import json import os import sys import tempfile import threading import types import unittest import asyncio from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, patch from app.agents.browser_use_bridge import ( BROWSER_USE_VERSION, CAMOUFOX_SETUP_COMMAND, CAMOUFOX_VERSION, CDP_PROXY_COMMAND, REMOTE_CDP_PROXY_PATH, REMOTE_STEALTH_SERVER_PATH, _create_e2b_desktop, _start_local_worker, _wait_for_external_cdp, _worker_environment, build_chrome_command, build_e2b_sandbox_kwargs, build_public_result, build_worker_config, cdp_proxy_script_source, run_browser_use_agent, translate_worker_event, worker_script_source, ) from app.agents.browser_use_worker import ( _browser_search_urls, _normalize_browser_links, _normalize_stealth_links, _playwright_cookies, ) import app.agents.cdp_proxy as cdp_proxy from app.agents.cdp_proxy import _authorized_and_sanitized MODEL = "nvidia/nemotron-3-ultra-550b-a55b:free" class _FakeFiles: def __init__(self): self.data = {} def write(self, path, data): self.data[path] = data return SimpleNamespace(path=path) class _FakeRemoteHandle: def __init__(self): self.killed = False def kill(self): self.killed = True return True class _FakeCommands: def __init__(self): self.calls = [] self.handles = [] def run(self, command, **kwargs): self.calls.append((command, kwargs)) handle = _FakeRemoteHandle() self.handles.append(handle) return handle class _FakeDesktop: def __init__(self): self.files = _FakeFiles() self.commands = _FakeCommands() self.stream = _FakeStream() self.killed = False def get_host(self, port): return f"{port}-sandbox.e2b.app" def kill(self): self.killed = True class _FakeStream: def __init__(self): self.started = False self.stopped = False def start(self, **kwargs): self.started = True def get_auth_key(self): return "visual-auth-key" def get_url(self, **kwargs): return "https://visual.example/session" def stop(self): self.stopped = True class _FakeSandbox: desktop = None create_kwargs = None @classmethod def create(cls, **kwargs): cls.create_kwargs = kwargs cls.desktop = _FakeDesktop() return cls.desktop class _FakeLocalProcess: def __init__(self, exit_code=0): self.returncode = exit_code self.terminated = False self.killed = False async def wait(self): return self.returncode def terminate(self): self.terminated = True self.returncode = -15 def kill(self): self.killed = True self.returncode = -9 def _successful_events_and_result(): events = [ { "type": "step", "step": 1, "url": "https://duckduckgo.com/?q=lulinha", "title": "Search", "actions": [{"navigate": {"url": "https://source-one.example/profile"}}], "next_goal": "Open the first direct source", }, { "type": "evidence", "record": { "claim": "Lulinha is Fábio Luís", "exact_quote": "Fábio Luís, conhecido como Lulinha", "url": "https://source-one.example/profile", "title": "Profile", "hostname": "source-one.example", }, }, { "type": "evidence", "record": { "claim": "A second source identifies the same person", "exact_quote": "Fábio Luís Lula da Silva, o Lulinha", "url": "https://source-two.example/story", "title": "Story", "hostname": "source-two.example", }, }, ] result = { "ok": True, "answer": "Lulinha is Fábio Luís. Sources are linked below.", "agent_success": True, "is_done": True, "urls": [ "https://duckduckgo.com/?q=lulinha", "https://source-one.example/profile", "https://source-two.example/story", ], "steps": 3, "duration_seconds": 9.2, "evidence": [item["record"] for item in events if item["type"] == "evidence"], "model": MODEL, } return events, result class _FakeResponse: status_code = 200 def json(self): return { "webSocketDebuggerUrl": ( "ws://127.0.0.1:9222/devtools/browser/internal-browser-id" ) } class _FakeHttpClient: calls = [] def __init__(self, *args, **kwargs): pass async def __aenter__(self): return self async def __aexit__(self, *args): return False async def get(self, url, headers): self.calls.append((url, headers)) return _FakeResponse() class BrowserUseBridgeTests(unittest.IsolatedAsyncioTestCase): def test_host_worker_is_standalone_remote_cdp_code(self): source = worker_script_source() compile(source, "", "exec") self.assertIn("from browser_use import", source) self.assertIn("record_evidence", source) self.assertIn("final_response_after_failure=True", source) self.assertNotIn(' "done",', source) self.assertNotIn("finish_research", source) self.assertIn("LANCER_CDP_URL", source) self.assertIn("x-lancer-cdp-token", source) self.assertIn("BrowserProfile", source) self.assertIn("block_ip_addresses=True", source) self.assertIn("MAX_PAGE_TEXT_CHARS", source) self.assertIn(".slice(0, limit)", source) self.assertIn("browser_search", source) self.assertIn("open_source", source) self.assertIn("_inspect_search_page", source) self.assertIn('"navigate",', source) self.assertIn('"search",', source) self.assertIn("enable_planning=False", source) self.assertIn("captcha_solver=True", source) self.assertIn('llm_kwargs["reasoning_effort"] = "low"', source) self.assertIn('llm_kwargs["reasoning_models"] = [worker_model]', source) self.assertIn('"max_retries": 3', source) self.assertIn("use_thinking=False", source) self.assertNotIn("SearXNG", source) self.assertNotIn("searxng", source.casefold()) self.assertIn("raw_page_text = await page.evaluate", source) self.assertIn("quote_normalized in page_text_normalized", source) self.assertNotIn("Fuzzy fallback", source) self.assertNotIn("match_ratio", source) self.assertNotIn("executable_path=", source) self.assertIn("Camoufox", source) self.assertIn("stealth_gateway_url", source) self.assertIn("captcha_solved", source) def test_camoufox_snapshot_links_and_cookies_are_sanitized(self): links = _normalize_stealth_links( { "elements": [ { "text": "Search link", "href": "https://search.brave.com/help", }, { "text": "Direct source", "href": "https://news.example.org/article", }, ] } ) self.assertEqual( links, [{"title": "Direct source", "url": "https://news.example.org/article"}], ) cookies = _playwright_cookies( [ { "name": "clearance", "value": "ok", "domain": ".example.org", "path": "/", "sameSite": "lax", "unsupported": "discard-me", } ] ) self.assertEqual(cookies[0]["sameSite"], "Lax") self.assertNotIn("unsupported", cookies[0]) def test_live_browser_search_filters_serp_links_and_unwraps_targets(self): engines = dict(_browser_search_urls("quem é lulinha?")) self.assertIn("search.brave.com", engines["Brave Search"]) self.assertNotIn("google.com", " ".join(engines.values())) links = _normalize_browser_links( [ {"title": "Search navigation", "url": "https://search.brave.com/help"}, { "title": "Direct article", "url": "https://noticias.example.org/reportagem", }, { "title": "Wrapped article", "url": ( "https://duckduckgo.com/l/?uddg=" "https%3A%2F%2Fsecond.example.com%2Fprofile" ), }, { "title": "Yahoo wrapped article", "url": ( "https://r.search.yahoo.com/_ylt=x/RV=2/RU=" "https%3A%2F%2Fthird.example.net%2Fbio/RK=2/RS=x" ), }, ] ) self.assertEqual( [item["url"] for item in links], [ "https://noticias.example.org/reportagem", "https://second.example.com/profile", "https://third.example.net/bio", ], ) def test_proxy_is_dependency_free_authenticated_and_rewrites_host(self): source = cdp_proxy_script_source() compile(source, "", "exec") request = ( b"GET /json/version HTTP/1.1\r\n" b"Host: 9223-public.e2b.app\r\n" b"X-Lancer-CDP-Token: correct-token\r\n" b"Connection: Upgrade\r\n\r\n" ) sanitized = _authorized_and_sanitized(request, "correct-token") self.assertIsNotNone(sanitized) self.assertIn(b"Host: 127.0.0.1:9222", sanitized) self.assertNotIn(b"correct-token", sanitized) self.assertIsNone(_authorized_and_sanitized(request, "wrong-token")) async def test_proxy_tunnels_websocket_bytes_after_authenticated_upgrade(self): captured = {} upstream_done = asyncio.Event() proxy_done = asyncio.Event() async def fake_chrome(reader, writer): try: captured["headers"] = await reader.readuntil(b"\r\n\r\n") writer.write( b"HTTP/1.1 101 Switching Protocols\r\n" b"Upgrade: websocket\r\n" b"Connection: Upgrade\r\n\r\n" ) await writer.drain() payload = await reader.readexactly(4) writer.write(payload.upper()) await writer.drain() finally: writer.close() await writer.wait_closed() upstream_done.set() async def proxy_connection(reader, writer): try: await cdp_proxy._handle_client(reader, writer) finally: proxy_done.set() upstream = await asyncio.start_server(fake_chrome, "127.0.0.1", 0) upstream_port = upstream.sockets[0].getsockname()[1] with ( patch.object(cdp_proxy, "TOKEN", "correct-token"), patch.object(cdp_proxy, "UPSTREAM_PORT", upstream_port), ): proxy = await asyncio.start_server( proxy_connection, "127.0.0.1", 0, ) proxy_port = proxy.sockets[0].getsockname()[1] reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port) writer.write( b"GET /devtools/browser/id HTTP/1.1\r\n" b"Host: public.example\r\n" b"Upgrade: websocket\r\n" b"Connection: Upgrade\r\n" b"X-Lancer-CDP-Token: correct-token\r\n\r\n" ) await writer.drain() response = await reader.readuntil(b"\r\n\r\n") self.assertIn(b"101 Switching Protocols", response) writer.write(b"ping") await writer.drain() self.assertEqual(await reader.readexactly(4), b"PING") writer.close() await writer.wait_closed() await asyncio.wait_for(upstream_done.wait(), timeout=2) await asyncio.wait_for(proxy_done.wait(), timeout=2) proxy.close() await proxy.wait_closed() upstream.close() await upstream.wait_closed() self.assertIn(f"Host: 127.0.0.1:{upstream_port}".encode(), captured["headers"]) self.assertNotIn(b"correct-token", captured["headers"]) def test_chrome_launch_is_remote_debuggable_and_headless_is_explicit(self): visual = build_chrome_command(visual=True) headless = build_chrome_command(visual=False) self.assertIn("--remote-debugging-port=9222", visual) self.assertIn("--remote-allow-origins=*", visual) self.assertNotIn("--headless=new", visual) self.assertIn("--headless=new", headless) def test_config_is_non_secret_and_uses_current_normal_depth_preset(self): config = build_worker_config( task="quem é lulinha?", url=None, visual=False, model=MODEL, ) serialized = json.dumps(config) self.assertEqual(config["minimum_sources"], 2) self.assertEqual(config["min_evidence_to_finish"], 2) self.assertTrue(config["headless"]) self.assertEqual(config["max_steps"], 18) self.assertEqual(config["llm_timeout"], 120) self.assertEqual(config["step_timeout"], 150) self.assertEqual(config["worker_timeout"], 600) self.assertNotIn("api_key", serialized.casefold()) self.assertNotIn("secret", serialized.casefold()) def test_prebuilt_e2b_template_is_optional_and_explicit(self): default_settings = SimpleNamespace(e2b_api_key="key") template_settings = SimpleNamespace( e2b_api_key="key", e2b_browser_template="lancer-browser-v1", ) self.assertNotIn("template", build_e2b_sandbox_kwargs(default_settings)) self.assertEqual( build_e2b_sandbox_kwargs(template_settings)["template"], "lancer-browser-v1", ) def test_worker_environment_does_not_copy_unrelated_host_secrets(self): with patch.dict( os.environ, { "E2B_API_KEY": "must-not-cross", "AWS_SECRET_ACCESS_KEY": "must-not-cross-either", }, ): environment = _worker_environment( openrouter_api_key="openrouter-key", cdp_url="wss://9223-sandbox.e2b.app/devtools/browser/id", cdp_token="cdp-token", runtime_dir=Path("/tmp/test-runtime"), ) self.assertNotIn("E2B_API_KEY", environment) self.assertNotIn("AWS_SECRET_ACCESS_KEY", environment) self.assertEqual(environment["OPENROUTER_API_KEY"], "openrouter-key") self.assertEqual(environment["LANCER_CDP_TOKEN"], "cdp-token") self.assertEqual(environment["PYTHON_DOTENV_DISABLED"], "1") self.assertEqual(environment["PYTHONNOUSERSITE"], "1") self.assertEqual(environment["HOME"], str(Path("/tmp/test-runtime") / "home")) self.assertNotIn("PYTHONPATH", environment) async def test_local_worker_starts_in_its_private_runtime_directory(self): fake_process = _FakeLocalProcess() with tempfile.TemporaryDirectory() as temp_dir: runtime = Path(temp_dir) config_path = runtime / "config.json" config_path.write_text("{}", encoding="utf-8") with patch( "app.agents.browser_use_bridge.asyncio.create_subprocess_exec", new=AsyncMock(return_value=fake_process), ) as spawn: await _start_local_worker( config_path=config_path, events_path=runtime / "events.jsonl", result_path=runtime / "result.json", log_path=runtime / "worker.log", environment={"PATH": os.environ.get("PATH", "")}, ) self.assertEqual(spawn.await_args.kwargs["cwd"], str(runtime)) async def test_cancelled_creation_kills_late_e2b_sandbox(self): started = threading.Event() release = threading.Event() created = _FakeDesktop() class SlowSandbox: @classmethod def create(cls, **kwargs): started.set() release.wait(timeout=2) return created task = asyncio.create_task(_create_e2b_desktop(SlowSandbox, timeout=720)) self.assertTrue(await asyncio.to_thread(started.wait, 1)) task.cancel() release.set() with self.assertRaises(asyncio.CancelledError): await task self.assertTrue(created.killed) async def test_external_cdp_probe_converts_internal_ws_to_public_wss(self): _FakeHttpClient.calls = [] desktop = _FakeDesktop() with patch( "app.agents.browser_use_bridge.httpx.AsyncClient", _FakeHttpClient, ): url = await _wait_for_external_cdp(desktop, "per-run-token") self.assertEqual( url, "wss://9223-sandbox.e2b.app/devtools/browser/internal-browser-id", ) request_url, headers = _FakeHttpClient.calls[0] self.assertEqual(request_url, "https://9223-sandbox.e2b.app/json/version") self.assertEqual(headers["x-lancer-cdp-token"], "per-run-token") def test_translation_exposes_only_accepted_evidence_as_facts(self): progress = {"evidence": [], "next_goals": [], "queries": []} events = translate_worker_event( { "type": "step", "step": 2, "url": "https://example.com", "title": "Example", "actions": [{"browser_search": {"query": "focused query"}}], "memory": "Unverified model memory must not appear as evidence", "next_goal": "Open the original source", }, progress, ) public_progress = next(item for item in events if item["type"] == "progress") self.assertEqual(public_progress["known_facts"], []) self.assertEqual(public_progress["last_queries"], ["focused query"]) self.assertEqual(public_progress["missing_points"], ["Open the original source"]) def test_public_result_never_promotes_search_or_visited_urls(self): config = build_worker_config( task="quem é alguém?", url=None, visual=False, model="test-model", ) public = build_public_result( { "ok": True, "answer": "Useful partial answer", "agent_success": True, "urls": [ "https://duckduckgo.com/?q=someone", "https://unverified.example/page", "https://verified.example/page", ], "evidence": [ { "claim": "Verified claim", "exact_quote": "A sufficiently long exact quotation", "url": "https://verified.example/page", }, { "claim": "Snippet claim", "exact_quote": "A search snippet must never count", "url": "https://duckduckgo.com/?q=someone", }, ], }, config, ) self.assertEqual(public["links"], ["https://verified.example/page"]) self.assertFalse(public["success"]) self.assertIn("1/2", public["content"]) self.assertIn("https://unverified.example/page", public["visited_links"]) self.assertEqual(public["verification_status"], "partial") def test_public_result_without_recorded_evidence_is_never_verified(self): config = build_worker_config( task="research a person", url=None, visual=False, model="test-model", ) public = build_public_result( { "ok": True, "answer": "A useful evaluator draft", "agent_success": True, "urls": ["https://visited.example/profile"], "evidence": [], }, config, ) self.assertFalse(public["success"]) self.assertEqual(public["links"], []) self.assertEqual(public["evidence"], []) self.assertEqual(public["verification_status"], "partial") self.assertIn("0/1", public["content"]) self.assertNotIn("Citação validada pelo Evaluator", worker_script_source()) def test_subdomains_of_one_publisher_are_not_independent_sources(self): config = build_worker_config( task="quem é alguém?", url=None, visual=False, model="test-model", ) public = build_public_result( { "ok": True, "answer": "Answer", "agent_success": True, "evidence": [ { "claim": "Claim one", "exact_quote": "A sufficiently long literal quote one", "url": "https://pt.wikipedia.org/wiki/Person", }, { "claim": "Claim two", "exact_quote": "A sufficiently long literal quote two", "url": "https://en.wikipedia.org/wiki/Person", }, ], }, config, ) self.assertEqual(public["verified_sources"], 1) self.assertFalse(public["success"]) async def test_full_bridge_runs_host_browser_use_with_camoufox_in_e2b(self): fake_module = types.ModuleType("e2b_desktop") fake_module.Sandbox = _FakeSandbox settings = SimpleNamespace( e2b_api_key="e2b-test-key", openrouter_api_key="openrouter-test-key", llm_model=MODEL, ) captured = {} async def fake_start_local_worker(**kwargs): captured["environment"] = dict(kwargs["environment"]) captured["config"] = kwargs["config_path"].read_text(encoding="utf-8") events, result = _successful_events_and_result() kwargs["events_path"].write_text( "\n".join(json.dumps(item) for item in events) + "\n", encoding="utf-8", ) kwargs["result_path"].write_text(json.dumps(result), encoding="utf-8") return _FakeLocalProcess() _FakeSandbox.desktop = None with ( patch.dict(sys.modules, {"e2b_desktop": fake_module}), patch("app.agents.browser_use_bridge.get_settings", return_value=settings), patch( "app.agents.browser_use_bridge.installed_browser_use_version", return_value=BROWSER_USE_VERSION, ), patch( "app.agents.browser_use_bridge._wait_for_external_cdp", new=AsyncMock( return_value=( "wss://9223-sandbox.e2b.app/devtools/browser/test-browser" ) ), ), patch( "app.agents.browser_use_bridge._wait_for_external_stealth_browser", new=AsyncMock(return_value="https://8766-sandbox.e2b.app"), ), patch( "app.agents.browser_use_bridge._start_local_worker", side_effect=fake_start_local_worker, ), ): events = [ event async for event in run_browser_use_agent( "quem é lulinha?", visual=False, ) ] desktop = _FakeSandbox.desktop commands = [command for command, _ in desktop.commands.calls] self.assertTrue(any("google-chrome" in command for command in commands)) self.assertIn(CDP_PROXY_COMMAND, commands) self.assertIn(CAMOUFOX_SETUP_COMMAND, commands) self.assertTrue(any(f"camoufox=={CAMOUFOX_VERSION}" in command for command in commands)) self.assertFalse(any("browser-use==" in command for command in commands)) self.assertNotIn("openrouter-test-key", " ".join(commands)) self.assertIn(REMOTE_CDP_PROXY_PATH, desktop.files.data) self.assertIn(REMOTE_STEALTH_SERVER_PATH, desktop.files.data) self.assertTrue(desktop.killed) background_handles = [ handle for (_, kwargs), handle in zip( desktop.commands.calls, desktop.commands.handles, ) if kwargs.get("background") ] self.assertTrue(background_handles) self.assertTrue(all(handle.killed for handle in background_handles)) # Regression: E2B rejects several mathematically valid reserved CIDRs # (including 0.0.0.0/8) with HTTP 400. Network policy must not prevent # the browser sandbox from being created. self.assertNotIn("network", _FakeSandbox.create_kwargs) self.assertNotIn("template", _FakeSandbox.create_kwargs) config_text = captured["config"] worker_environment = captured["environment"] self.assertNotIn("openrouter-test-key", config_text) self.assertNotIn("e2b-test-key", config_text) self.assertIn("https://8766-sandbox.e2b.app", config_text) self.assertEqual(worker_environment["OPENROUTER_API_KEY"], "openrouter-test-key") self.assertNotIn("E2B_API_KEY", worker_environment) result = next(item for item in events if item["type"] == "result") self.assertTrue(result["success"]) self.assertEqual(result["verified_sources"], 2) self.assertNotIn("https://duckduckgo.com/?q=lulinha", result["links"]) async def test_missing_pinned_host_runtime_fails_before_creating_e2b(self): settings = SimpleNamespace( e2b_api_key="e2b-test-key", openrouter_api_key="openrouter-test-key", llm_model=MODEL, ) _FakeSandbox.desktop = None with ( patch("app.agents.browser_use_bridge.get_settings", return_value=settings), patch( "app.agents.browser_use_bridge.installed_browser_use_version", return_value="", ), ): events = [ event async for event in run_browser_use_agent("research", visual=False) ] self.assertEqual(len(events), 1) self.assertEqual(events[0]["type"], "error") self.assertIn(f"browser-use=={BROWSER_USE_VERSION}", events[0]["message"]) self.assertIsNone(_FakeSandbox.desktop) async def test_runtime_failure_redacts_every_request_credential(self): fake_module = types.ModuleType("e2b_desktop") fake_module.Sandbox = _FakeSandbox settings = SimpleNamespace( e2b_api_key="e2b-test-key", openrouter_api_key="openrouter-test-key", llm_model=MODEL, ) failure = RuntimeError( "gateway rejected cdp-secret-token openrouter-test-key e2b-test-key" ) _FakeSandbox.desktop = None with ( patch.dict(sys.modules, {"e2b_desktop": fake_module}), patch("app.agents.browser_use_bridge.get_settings", return_value=settings), patch( "app.agents.browser_use_bridge.installed_browser_use_version", return_value=BROWSER_USE_VERSION, ), patch( "app.agents.browser_use_bridge.secrets.token_urlsafe", return_value="cdp-secret-token", ), patch( "app.agents.browser_use_bridge._wait_for_external_cdp", new=AsyncMock(side_effect=failure), ), self.assertLogs("app.agents.browser_use_bridge", level="ERROR") as logs, ): events = [ event async for event in run_browser_use_agent("research", visual=False) ] combined = json.dumps(events, ensure_ascii=False) + "\n" + "\n".join(logs.output) self.assertIn("[REDACTED]", combined) self.assertNotIn("cdp-secret-token", combined) self.assertNotIn("openrouter-test-key", combined) self.assertNotIn("e2b-test-key", combined) self.assertTrue(_FakeSandbox.desktop.killed) self.assertTrue( all(handle.killed for handle in _FakeSandbox.desktop.commands.handles) ) async def test_visual_failure_ends_stream_before_reporting_error(self): fake_module = types.ModuleType("e2b_desktop") fake_module.Sandbox = _FakeSandbox settings = SimpleNamespace( e2b_api_key="e2b-test-key", openrouter_api_key="openrouter-test-key", llm_model=MODEL, ) _FakeSandbox.desktop = None with ( patch.dict(sys.modules, {"e2b_desktop": fake_module}), patch("app.agents.browser_use_bridge.get_settings", return_value=settings), patch( "app.agents.browser_use_bridge.installed_browser_use_version", return_value=BROWSER_USE_VERSION, ), patch( "app.agents.browser_use_bridge._wait_for_external_cdp", new=AsyncMock( return_value="wss://9223-sandbox.e2b.app/devtools/browser/id" ), ), patch( "app.agents.browser_use_bridge._start_local_worker", new=AsyncMock(side_effect=RuntimeError("worker startup failed")), ), ): events = [ event async for event in run_browser_use_agent("research", visual=True) ] event_types = [event["type"] for event in events] self.assertIn("stream", event_types) self.assertLess(event_types.index("stream_end"), event_types.index("error")) self.assertTrue(_FakeSandbox.desktop.stream.stopped) if __name__ == "__main__": unittest.main()