diff --git a/bin/cloakserve b/bin/cloakserve index d1719e34..2fd083db 100755 --- a/bin/cloakserve +++ b/bin/cloakserve @@ -319,7 +319,14 @@ class ChromePool: # Check if already running (including default fast-path) if seed_key in self._processes: proc = self._processes[seed_key] - if proc.process.poll() is None: + # poll() alone is not enough: during an identity-lease rebuild a + # client closes Chrome and immediately relaunches on the same + # seed, and Chrome tears down its DevTools/CDP socket seconds + # before the OS process actually exits. So a poll()-alive process + # can already be refusing CDP connections — returning it hands the + # client a corpse. Verify the CDP port is still accepting + # connections (cheap, sub-second, local) before reusing it. + if proc.process.poll() is None and await self._cdp_port_alive(proc.cdp_port): if seed_key in self._idle_tasks: self._schedule_idle_cleanup(seed_key) if any([extra_args, timezone, locale, proxy, geoip]): @@ -330,7 +337,14 @@ class ChromePool: proc.timezone, proc.locale, proc.proxy, ) return proc - # Dead — clean up + # Dead: the process exited, or it is poll()-alive but its CDP + # port is no longer accepting connections (a dying corpse). Evict + # and fall through to a fresh launch. + logger.info( + "Seed %s pooled process is no longer serving CDP (port %d) — " + "evicting and relaunching", + seed_key, proc.cdp_port, + ) await self._cleanup_process(seed_key) # Resolve geoip if requested @@ -409,6 +423,19 @@ class ChromePool: logger.info("Chrome ready (seed=%s, port=%d, pid=%d)", actual_seed, port, process.pid) return cp + def _pooled_key(self, cp: ChromeProcess) -> str | None: + """Return the pool key currently mapping to cp. + + Returns None when cp is no longer the pooled process for its seed — + i.e. it was already evicted, or a concurrent request relaunched a fresh + process in its place. Callers evict by this key so a stale corpse is + removed without ever nuking a healthy relaunched process. + """ + for key, proc in self._processes.items(): + if proc is cp: + return key + return None + async def _cleanup_process(self, key: str) -> None: """Terminate a Chrome process and clean up.""" self._cancel_idle_cleanup(key) @@ -464,6 +491,31 @@ class ChromePool: finally: await session.close() + @staticmethod + async def _cdp_port_alive(port: int, timeout: float = 0.5) -> bool: + """Cheap liveness probe: is the CDP port still accepting TCP connections? + + Chrome closes its DevTools listening socket seconds before the OS + process exits, so process.poll() alone reports a dying Chrome as alive. + A single sub-second local TCP connect distinguishes a genuinely-serving + CDP endpoint from a poll()-alive corpse, keeping the warm-hit happy path + fast (a live listener completes the handshake in well under a + millisecond; a dead port refuses immediately). + """ + try: + _reader, writer = await asyncio.wait_for( + asyncio.open_connection("127.0.0.1", port), + timeout=timeout, + ) + except (OSError, asyncio.TimeoutError): + return False + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + return True + # --------------------------------------------------------------------------- # Query param parsing @@ -556,25 +608,37 @@ async def handle_json_version(request: web.Request) -> web.Response: pool: ChromePool = request.app["pool"] params = parse_connection_params(request.query_string) - cp = await pool.get_or_launch( - seed=params["seed"], - extra_args=params["extra_args"] or None, - timezone=params["timezone"], - locale=params["locale"], - proxy=params["proxy"], - geoip=params["geoip"], - ) - - try: - async with aiohttp.ClientSession() as session: - async with session.get( - f"http://127.0.0.1:{cp.cdp_port}/json/version", - timeout=aiohttp.ClientTimeout(total=5), - ) as resp: - data = await resp.json() - except Exception as exc: - logger.error("Failed to reach Chrome CDP (port %d): %s", cp.cdp_port, exc) - return web.json_response({"error": "CDP endpoint unreachable"}, status=502) + # Defense-in-depth: get_or_launch already probes CDP liveness before reuse, + # but a process can still die in the window between that probe and this + # fetch. If the fetch fails with a connection error, evict the stale pool + # entry and relaunch exactly once so a corpse becomes a single cold + # relaunch instead of a persistent 502. + data = None + for attempt in range(2): + cp = await pool.get_or_launch( + seed=params["seed"], + extra_args=params["extra_args"] or None, + timezone=params["timezone"], + locale=params["locale"], + proxy=params["proxy"], + geoip=params["geoip"], + ) + try: + async with aiohttp.ClientSession() as session: + async with session.get( + f"http://127.0.0.1:{cp.cdp_port}/json/version", + timeout=aiohttp.ClientTimeout(total=5), + ) as resp: + data = await resp.json() + break + except Exception as exc: + logger.error("Failed to reach Chrome CDP (port %d): %s", cp.cdp_port, exc) + if attempt == 0: + stale_key = pool._pooled_key(cp) + if stale_key is not None: + await pool._cleanup_process(stale_key) + continue + return web.json_response({"error": "CDP endpoint unreachable"}, status=502) # Rewrite webSocketDebuggerUrl to route through our multiplexer host = _external_host(request) @@ -598,25 +662,35 @@ async def handle_json_list(request: web.Request) -> web.Response: pool: ChromePool = request.app["pool"] params = parse_connection_params(request.query_string) - cp = await pool.get_or_launch( - seed=params["seed"], - extra_args=params["extra_args"] or None, - timezone=params["timezone"], - locale=params["locale"], - proxy=params["proxy"], - geoip=params["geoip"], - ) - - try: - async with aiohttp.ClientSession() as session: - async with session.get( - f"http://127.0.0.1:{cp.cdp_port}/json/list", - timeout=aiohttp.ClientTimeout(total=5), - ) as resp: - data = await resp.json() - except Exception as exc: - logger.error("Failed to reach Chrome CDP (port %d): %s", cp.cdp_port, exc) - return web.json_response({"error": "CDP endpoint unreachable"}, status=502) + # Defense-in-depth: evict + relaunch once if a pooled process dies in the + # window between get_or_launch's liveness probe and this fetch (see + # handle_json_version for the rationale). + data = None + for attempt in range(2): + cp = await pool.get_or_launch( + seed=params["seed"], + extra_args=params["extra_args"] or None, + timezone=params["timezone"], + locale=params["locale"], + proxy=params["proxy"], + geoip=params["geoip"], + ) + try: + async with aiohttp.ClientSession() as session: + async with session.get( + f"http://127.0.0.1:{cp.cdp_port}/json/list", + timeout=aiohttp.ClientTimeout(total=5), + ) as resp: + data = await resp.json() + break + except Exception as exc: + logger.error("Failed to reach Chrome CDP (port %d): %s", cp.cdp_port, exc) + if attempt == 0: + stale_key = pool._pooled_key(cp) + if stale_key is not None: + await pool._cleanup_process(stale_key) + continue + return web.json_response({"error": "CDP endpoint unreachable"}, status=502) host = _external_host(request) scheme = _ws_scheme(request) diff --git a/tests/test_cloakserve.py b/tests/test_cloakserve.py index 268b120d..1fbf7a10 100644 --- a/tests/test_cloakserve.py +++ b/tests/test_cloakserve.py @@ -472,6 +472,7 @@ def _track_process(self, pool, seed="seed1"): def _track_live_process(self, pool, seed="seed1"): pool._processes[seed] = SimpleNamespace( process=SimpleNamespace(poll=lambda: None), + cdp_port=5100, ) def test_connect_increments(self): @@ -571,6 +572,11 @@ async def run(): pool = self._make_pool(idle_timeout=1.0) self._track_live_process(pool) + async def _alive(_port): + return True + + pool._cdp_port_alive = _alive + pool.connect("seed1") pool.disconnect("seed1") first_task = pool._idle_tasks["seed1"] @@ -676,3 +682,294 @@ def test_refuses_traversal_path(self, tmp_path): pool._safe_rmtree(traversal) assert victim.exists(), "Traversal path must not be deleted" + + +# --------------------------------------------------------------------------- +# ADR-0163 stage 0 — poll()-alive but CDP-dead corpse eviction +# +# During an olabrowser identity-lease rebuild, a client closes Chrome and +# immediately relaunches on the SAME fingerprint seed. Chrome tears down its +# DevTools/CDP socket seconds before the OS process exits, so process.poll() +# still reports the dying process as alive. The pool must NOT hand out that +# corpse — it must probe the CDP port, evict, and relaunch. +# --------------------------------------------------------------------------- + + +class TestGetOrLaunchCdpLiveness: + """get_or_launch must not reuse a poll()-alive process whose CDP port is dead.""" + + def _make_pool(self, data_dir): + return ChromePool( + binary="/fake/chrome", + global_args=[], + headless=True, + data_dir=data_dir, + ) + + def test_poll_alive_but_cdp_dead_is_evicted_and_relaunched(self, tmp_path, monkeypatch): + async def run(): + pool = self._make_pool(str(tmp_path)) + corpse = SimpleNamespace( + process=SimpleNamespace(poll=lambda: None), # poll() says "alive" + cdp_port=5100, + timezone=None, + locale=None, + proxy=None, + ) + pool._processes["seed1"] = corpse + + # CDP port refuses connections — the corpse is not really serving. + async def dead_probe(_port): + return False + + pool._cdp_port_alive = dead_probe + + evicted = [] + + async def fake_cleanup(key): + evicted.append(key) + pool._processes.pop(key, None) + + pool._cleanup_process = fake_cleanup + + # Stub the fresh launch so no real Chrome / arg-building is needed. + monkeypatch.setattr(_mod, "build_args", lambda **_kw: ["--stub"]) + monkeypatch.setattr(_mod, "_resolve_webrtc_args", lambda args, _proxy: args) + fresh_proc = SimpleNamespace(pid=4321, poll=lambda: None) + monkeypatch.setattr(_mod.subprocess, "Popen", lambda *_a, **_k: fresh_proc) + + async def fake_wait(_port, timeout=10.0): + return True + + pool._wait_for_cdp = fake_wait + + cp = await pool.get_or_launch("seed1") + + assert evicted == ["seed1"], "poll()-alive but CDP-dead corpse must be evicted" + assert cp is not corpse, "must not hand out the corpse" + assert cp.process is fresh_proc, "must relaunch a fresh process" + assert pool._processes["seed1"] is cp, "fresh process must be pooled under the seed" + + asyncio.run(run()) + + def test_poll_alive_and_cdp_alive_returns_same_process_fast(self, tmp_path, monkeypatch): + async def run(): + pool = self._make_pool(str(tmp_path)) + live = SimpleNamespace( + process=SimpleNamespace(poll=lambda: None), + cdp_port=5100, + timezone=None, + locale=None, + proxy=None, + ) + pool._processes["seed1"] = live + + probed = [] + + async def alive_probe(port): + probed.append(port) + return True + + pool._cdp_port_alive = alive_probe + + # A genuinely-live process must be returned after only the cheap + # probe — never relaunched. + def boom(*_a, **_k): + raise AssertionError("must not relaunch a live process") + + monkeypatch.setattr(_mod.subprocess, "Popen", boom) + + cp = await pool.get_or_launch("seed1") + + assert cp is live, "live process must be reused, not relaunched" + assert probed == [5100], "happy path must run exactly one cheap CDP probe" + + asyncio.run(run()) + + def test_process_exited_skips_probe_and_relaunches(self, tmp_path, monkeypatch): + async def run(): + pool = self._make_pool(str(tmp_path)) + dead = SimpleNamespace( + process=SimpleNamespace(poll=lambda: 0), # poll() says "exited" + cdp_port=5100, + timezone=None, + locale=None, + proxy=None, + ) + pool._processes["seed1"] = dead + + # If poll() already reports exit, the probe must be short-circuited. + async def probe_should_not_run(_port): + raise AssertionError("probe must be skipped when poll() reports exit") + + pool._cdp_port_alive = probe_should_not_run + + evicted = [] + + async def fake_cleanup(key): + evicted.append(key) + pool._processes.pop(key, None) + + pool._cleanup_process = fake_cleanup + + monkeypatch.setattr(_mod, "build_args", lambda **_kw: ["--stub"]) + monkeypatch.setattr(_mod, "_resolve_webrtc_args", lambda args, _proxy: args) + fresh_proc = SimpleNamespace(pid=4322, poll=lambda: None) + monkeypatch.setattr(_mod.subprocess, "Popen", lambda *_a, **_k: fresh_proc) + + async def fake_wait(_port, timeout=10.0): + return True + + pool._wait_for_cdp = fake_wait + + cp = await pool.get_or_launch("seed1") + + assert evicted == ["seed1"] + assert cp.process is fresh_proc + + asyncio.run(run()) + + +class _RaiseCtx: + """Async context manager whose __aenter__ mimics a refused CDP connection.""" + + async def __aenter__(self): + raise ConnectionRefusedError("connection refused") + + async def __aexit__(self, *_exc): + return None + + +class _JsonCtx: + def __init__(self, data): + self._data = data + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return None + + async def json(self): + return self._data + + +class _FakeSessionCM: + """Stand-in for `aiohttp.ClientSession()` that routes GETs by port.""" + + def __init__(self, resolver): + self._resolver = resolver + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return None + + def get(self, url, **_kwargs): + return self._resolver(url) + + +class _FakePool: + """Minimal pool: hands out a cp per get_or_launch call, tracks evictions.""" + + def __init__(self, ports): + self._ports = list(ports) + self.launches = 0 + self.evicted = [] + self._default_seed = None + self._processes = {} + + async def get_or_launch(self, **_kwargs): + port = self._ports[min(self.launches, len(self._ports) - 1)] + self.launches += 1 + cp = SimpleNamespace(cdp_port=port) + self._processes["seed1"] = cp + return cp + + async def _cleanup_process(self, key): + self.evicted.append(key) + self._processes.pop(key, None) + + def _pooled_key(self, cp): + for key, proc in self._processes.items(): + if proc is cp: + return key + return None + + +class _HandlerRequest: + def __init__(self, pool, query_string="fingerprint=seed1"): + self.headers = {"Host": "internal:9222"} + self.query_string = query_string + self.scheme = "http" + self.app = {"port": 9222, "pool": pool} + + +class TestJsonHandlerCorpseRecovery: + """handle_json_version / handle_json_list must evict + retry once on a dead port.""" + + def _patch_session(self, monkeypatch, resolver): + monkeypatch.setattr( + _mod.aiohttp, + "ClientSession", + lambda *_a, **_k: _FakeSessionCM(resolver), + ) + + def test_json_version_recovers_from_dead_port_without_502(self, monkeypatch): + pool = _FakePool(ports=[5100, 5200]) # first dead, relaunch alive + + def resolver(url): + if ":5100/" in url: + return _RaiseCtx() + return _JsonCtx({ + "webSocketDebuggerUrl": "ws://127.0.0.1:5200/devtools/browser/guid", + }) + + self._patch_session(monkeypatch, resolver) + + response = asyncio.run(_mod.handle_json_version(_HandlerRequest(pool))) + + assert response.status == 200, "dead corpse must not surface as a 502" + assert pool.launches == 2, "must relaunch exactly once" + assert pool.evicted == ["seed1"], "stale pool entry must be evicted before retry" + payload = json.loads(response.text) + assert payload["webSocketDebuggerUrl"] == ( + "ws://internal:9222/fingerprint/seed1/devtools/browser/guid" + ) + + def test_json_version_persistent_dead_port_502_after_single_retry(self, monkeypatch): + pool = _FakePool(ports=[5100, 5100]) # both dead + + def resolver(url): + return _RaiseCtx() + + self._patch_session(monkeypatch, resolver) + + response = asyncio.run(_mod.handle_json_version(_HandlerRequest(pool))) + + assert response.status == 502, "persistent dead port still returns 502" + assert pool.launches == 2, "retry must happen exactly once (initial + 1)" + assert pool.evicted == ["seed1"], "evicted exactly once before the single retry" + + def test_json_list_recovers_from_dead_port_without_502(self, monkeypatch): + pool = _FakePool(ports=[5100, 5200]) + + def resolver(url): + if ":5100/" in url: + return _RaiseCtx() + return _JsonCtx([ + {"webSocketDebuggerUrl": "ws://127.0.0.1:5200/devtools/page/pageguid"}, + ]) + + self._patch_session(monkeypatch, resolver) + + response = asyncio.run(_mod.handle_json_list(_HandlerRequest(pool))) + + assert response.status == 200 + assert pool.launches == 2 + assert pool.evicted == ["seed1"] + payload = json.loads(response.text) + assert payload[0]["webSocketDebuggerUrl"] == ( + "ws://internal:9222/fingerprint/seed1/devtools/page/pageguid" + )