From c9e4f58353190ebb7e610d6c201daa97b08217b5 Mon Sep 17 00:00:00 2001 From: CloakHQ Date: Thu, 2 Apr 2026 18:34:17 +0200 Subject: [PATCH] feat: rewrite cloakserve as CDP multiplexer with per-connection fingerprint seeds Spawns a separate Chrome process per unique fingerprint seed, all behind a single port (9222). Clients specify seeds and fingerprint params via query string on the CDP URL: connect_over_cdp("http://host:9222?fingerprint=12345&timezone=Asia/Tokyo") Supports all --fingerprint-* flags as query params, geoip=true for auto timezone/locale from proxy IP, and proxy= for per-process proxies. - Rewrite bin/cloakserve from 57-line wrapper to aiohttp CDP multiplexer - Add ChromePool with per-seed process management and port allocation - Bidirectional WebSocket proxy for CDP traffic - URL rewriting for /json/version, /json/list, and WS paths - Rename _build_args -> build_args, _maybe_resolve_geoip -> maybe_resolve_geoip - Add aiohttp + websockets to serve optional deps - Dockerfile installs .[serve] extras - Add 20 unit tests for cloakserve (param parsing, CLI args, URL rewriting) --- Dockerfile | 2 +- README.md | 23 +- bin/cloakserve | 573 +++++++++++++++++++++++++++++-- cloakbrowser/__init__.py | 4 +- cloakbrowser/browser.py | 22 +- pyproject.toml | 1 + tests/test_build_args.py | 30 +- tests/test_cloakserve.py | 171 +++++++++ tests/test_geoip.py | 16 +- tests/test_launch_context.py | 2 +- tests/test_persistent_context.py | 22 +- tests/test_proxy.py | 16 +- 12 files changed, 795 insertions(+), 87 deletions(-) create mode 100644 tests/test_cloakserve.py diff --git a/Dockerfile b/Dockerfile index f0d954e..78c00e8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,7 +20,7 @@ WORKDIR /app # Python wrapper COPY pyproject.toml README.md LICENSE BINARY-LICENSE.md CHANGELOG.md ./ COPY cloakbrowser/ cloakbrowser/ -RUN pip install --no-cache-dir . +RUN pip install --no-cache-dir ".[serve]" # JS wrapper COPY js/ js/ diff --git a/README.md b/README.md index 787df9f..969ba3f 100644 --- a/README.md +++ b/README.md @@ -754,7 +754,28 @@ services: start_period: 10s ``` -Run multiple instances with different fingerprint seeds on different ports — each gets unique canvas noise, client rects, and other browser signals. Pass `--fingerprint=` in the command (e.g., `cloakserve --fingerprint=12345`). +**Per-connection fingerprint seeds** — run multiple browser identities from a single container. Each unique seed spawns a separate Chrome process with its own fingerprint: + +```python +# Each seed gets unique canvas noise, client rects, and other browser signals +b1 = pw.chromium.connect_over_cdp("http://localhost:9222?fingerprint=11111") +b2 = pw.chromium.connect_over_cdp("http://localhost:9222?fingerprint=22222") + +# Full identity control via query params +b3 = pw.chromium.connect_over_cdp( + "http://localhost:9222?fingerprint=33333" + "&timezone=Asia/Tokyo&locale=ja-JP&platform=macos" + "&hardware-concurrency=4&device-memory=8" +) + +# Auto-detect timezone/locale from proxy exit IP +b4 = pw.chromium.connect_over_cdp( + "http://localhost:9222?fingerprint=44444" + "&proxy=http://proxy:8080&geoip=true" +) +``` + +Supported query params: `fingerprint`, `timezone`, `locale`, `platform`, `platform-version`, `brand`, `brand-version`, `gpu-vendor`, `gpu-renderer`, `hardware-concurrency`, `device-memory`, `screen-width`, `screen-height`, `proxy`, `geoip`. Same seed reuses the same process. No seed = shared default process (backward compatible). **Persistent profiles** — mount a volume to keep cookies and sessions across container restarts: diff --git a/bin/cloakserve b/bin/cloakserve index b58d5ef..69f4380 100755 --- a/bin/cloakserve +++ b/bin/cloakserve @@ -1,31 +1,51 @@ #!/usr/bin/env python3 -"""Launch stealth Chromium as a CDP server for remote connections. +"""CDP multiplexer — per-connection fingerprint seeds for stealth Chromium. + +Spawns a separate Chrome process per unique fingerprint seed, routing CDP +connections through a single port. Each seed gets its own browser identity. Usage: - cloakserve # headless on port 9222 - cloakserve --headless=false # headed (uses Xvfb in Docker) - cloakserve --proxy-server=host:port # with proxy + cloakserve # default, backward compat + cloakserve --port=9222 # custom port -Connect from host: - playwright.chromium.connect_over_cdp("http://localhost:9222") +Client: + browser = pw.chromium.connect_over_cdp("http://host:9222?fingerprint=12345") + browser = pw.chromium.connect_over_cdp( + "http://host:9222?fingerprint=12345&timezone=America/New_York&locale=en-US" + ) """ -import signal + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import random +import shutil +import socket import subprocess import sys import time +from dataclasses import dataclass +from urllib.parse import parse_qs -from cloakbrowser.config import get_default_stealth_args +import aiohttp +from aiohttp import web + +from cloakbrowser.browser import build_args, maybe_resolve_geoip from cloakbrowser.download import ensure_binary -PORT = 9222 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("cloakserve") -binary = ensure_binary() - -chrome_args = [ - binary, - f"--remote-debugging-port={PORT}", - "--remote-debugging-address=0.0.0.0", - # Sane defaults for running Chrome directly (outside Playwright) +# Args for running Chrome directly (outside Playwright). +# Playwright normally adds its own version of these. +BASE_CHROME_ARGS = [ "--no-first-run", "--no-default-browser-check", "--disable-dev-shm-usage", @@ -33,24 +53,517 @@ chrome_args = [ "--disable-popup-blocking", "--disable-background-networking", "--metrics-recording-only", - # GPU blocklist bypass: Chromium blocks WebGL on software GPUs in - # Docker/Xvfb. Without this, WebGL vendor/renderer spoofing fails. #58 "--ignore-gpu-blocklist", -] + get_default_stealth_args() + sys.argv[1:] +] -chrome = subprocess.Popen(chrome_args) - -time.sleep(2) - -print(f"CloakBrowser CDP server ready on port {PORT}", flush=True) +BASE_CDP_PORT = 5100 -def cleanup(sig, frame): - chrome.terminate() - sys.exit(0) +# --------------------------------------------------------------------------- +# ChromeProcess — one running Chrome instance +# --------------------------------------------------------------------------- + +@dataclass +class ChromeProcess: + seed: str + process: subprocess.Popen + cdp_port: int + user_data_dir: str -signal.signal(signal.SIGTERM, cleanup) -signal.signal(signal.SIGINT, cleanup) +# --------------------------------------------------------------------------- +# ChromePool — manages multiple Chrome processes keyed by seed +# --------------------------------------------------------------------------- -chrome.wait() +class ChromePool: + def __init__( + self, + binary: str, + global_args: list[str], + headless: bool, + ): + self._binary = binary + self._global_args = global_args + self._headless = headless + self._processes: dict[str, ChromeProcess] = {} + self._default: ChromeProcess | None = None + self._locks: dict[str, asyncio.Lock] = {} + self._next_port = BASE_CDP_PORT + + def _get_lock(self, seed: str) -> asyncio.Lock: + if seed not in self._locks: + self._locks[seed] = asyncio.Lock() + return self._locks[seed] + + def _allocate_port(self) -> int: + """Find a free port starting from _next_port.""" + for _ in range(100): + port = self._next_port + self._next_port += 1 + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", port)) + return port + except OSError: + continue + raise RuntimeError("No free ports available for Chrome CDP") + + async def get_or_launch( + self, + seed: str | None, + extra_args: list[str] | None = None, + timezone: str | None = None, + locale: str | None = None, + proxy: str | None = None, + geoip: bool = False, + ) -> ChromeProcess: + """Get existing or launch new Chrome process for a seed.""" + # No seed = default shared process + if seed is None: + seed_key = "__default__" + actual_seed = str(random.randint(10000, 99999)) + else: + seed_key = seed + actual_seed = seed + + lock = self._get_lock(seed_key) + async with lock: + # 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: + if any([extra_args, timezone, locale, proxy, geoip]): + logger.warning( + "Seed %s already running (port %d) — " + "ignoring new params (first-launch wins)", + seed_key, proc.cdp_port, + ) + return proc + # Dead — clean up + await self._cleanup_process(seed_key) + + # Resolve geoip if requested + if geoip and proxy: + timezone, locale = maybe_resolve_geoip(True, proxy, timezone, locale) + + # Build Chrome args via shared logic + fp_extra = [f"--fingerprint={actual_seed}"] + if extra_args: + fp_extra.extend(extra_args) + if proxy: + fp_extra.append(f"--proxy-server={proxy}") + + chrome_args = build_args( + stealth_args=True, + extra_args=fp_extra, + timezone=timezone, + locale=locale, + headless=self._headless, + ) + + # Allocate port and user data dir + port = self._allocate_port() + user_data_dir = f"/tmp/cloakserve-{seed_key}" + os.makedirs(user_data_dir, exist_ok=True) + + full_args = ( + [self._binary] + + BASE_CHROME_ARGS + + chrome_args + + self._global_args + + [ + f"--remote-debugging-port={port}", + "--remote-debugging-address=127.0.0.1", + f"--user-data-dir={user_data_dir}", + ] + ) + + logger.info("Launching Chrome (seed=%s, port=%d)", actual_seed, port) + process = subprocess.Popen( + full_args, + stdout=subprocess.DEVNULL, + ) + + # Wait for CDP to be ready + if not await self._wait_for_cdp(port): + process.kill() + await asyncio.to_thread(process.wait, timeout=5) + await asyncio.to_thread(shutil.rmtree, user_data_dir, True) + raise web.HTTPBadGateway( + text=json.dumps({"error": "Chrome failed to start"}), + content_type="application/json", + ) + + cp = ChromeProcess( + seed=actual_seed, + process=process, + cdp_port=port, + user_data_dir=user_data_dir, + ) + self._processes[seed_key] = cp + + if seed is None: + self._default = cp + + logger.info("Chrome ready (seed=%s, port=%d, pid=%d)", actual_seed, port, process.pid) + return cp + + async def _cleanup_process(self, key: str) -> None: + """Terminate a Chrome process and clean up.""" + proc = self._processes.pop(key, None) + if not proc: + return + if proc.process.poll() is None: + proc.process.terminate() + try: + await asyncio.to_thread(proc.process.wait, timeout=5) + except subprocess.TimeoutExpired: + proc.process.kill() + # Clean up user data dir (can be slow for large profiles) + await asyncio.to_thread(shutil.rmtree, proc.user_data_dir, True) + if self._default is proc: + self._default = None + self._locks.pop(key, None) + + async def shutdown(self) -> None: + """Terminate all Chrome processes.""" + for key in list(self._processes.keys()): + await self._cleanup_process(key) + logger.info("All Chrome processes terminated") + + @staticmethod + async def _wait_for_cdp(port: int, timeout: float = 10.0) -> bool: + """Poll Chrome's /json/version until ready.""" + deadline = time.monotonic() + timeout + delay = 0.1 + session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=1) + ) + try: + while time.monotonic() < deadline: + try: + async with session.get( + f"http://127.0.0.1:{port}/json/version" + ) as resp: + if resp.status == 200: + return True + except Exception: + pass + await asyncio.sleep(delay) + delay = min(delay * 2, 1.0) + return False + finally: + await session.close() + + +# --------------------------------------------------------------------------- +# Query param parsing +# --------------------------------------------------------------------------- + +# Params that need special handling (not simple --fingerprint-{name}= mapping) +SPECIAL_PARAMS = {"fingerprint", "proxy", "geoip", "locale", "timezone"} + + +def parse_connection_params(query_string: str) -> dict: + """Parse query params into connection config.""" + qs = parse_qs(query_string, keep_blank_values=False) + + result: dict = { + "seed": None, + "timezone": None, + "locale": None, + "proxy": None, + "geoip": False, + "extra_args": [], + } + + for key, values in qs.items(): + val = values[0] + if key == "fingerprint": + result["seed"] = val + elif key == "timezone": + result["timezone"] = val + elif key == "locale": + result["locale"] = val + elif key == "proxy": + result["proxy"] = val + elif key == "geoip": + result["geoip"] = val.lower() in ("true", "1", "yes") + elif key not in SPECIAL_PARAMS: + # Generic fingerprint param: map to --fingerprint-{key}={val} + result["extra_args"].append(f"--fingerprint-{key}={val}") + + return result + + +# --------------------------------------------------------------------------- +# HTTP handlers +# --------------------------------------------------------------------------- + +def _ws_scheme(request: web.Request) -> str: + """Return 'wss' if client connected via HTTPS (e.g. TLS-terminating proxy), else 'ws'.""" + proto = request.headers.get("X-Forwarded-Proto", request.scheme) + return "wss" if proto == "https" else "ws" + + +async def handle_root(request: web.Request) -> web.Response: + """Health check / info.""" + pool: ChromePool = request.app["pool"] + alive = sum(1 for p in pool._processes.values() if p.process.poll() is None) + return web.json_response({ + "status": "ok", + "processes": alive, + }) + + +async def handle_json_version(request: web.Request) -> web.Response: + """Proxy /json/version with optional per-seed routing.""" + 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) + + # Rewrite webSocketDebuggerUrl to route through our multiplexer + host = request.headers.get("Host", f"localhost:{request.app['port']}") + seed_key = params["seed"] + if seed_key: + ws_path = f"fingerprint/{seed_key}/devtools/browser" + else: + ws_path = "devtools/browser" + + # Extract the browser GUID from Chrome's original URL + orig_ws = data.get("webSocketDebuggerUrl", "") + guid = orig_ws.rsplit("/", 1)[-1] if "/devtools/" in orig_ws else "" + + scheme = _ws_scheme(request) + data["webSocketDebuggerUrl"] = f"{scheme}://{host}/{ws_path}/{guid}" + return web.json_response(data) + + +async def handle_json_list(request: web.Request) -> web.Response: + """Proxy /json/list with per-seed routing. Rewrites all entries.""" + 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) + + host = request.headers.get("Host", f"localhost:{request.app['port']}") + scheme = _ws_scheme(request) + seed_key = params["seed"] + + for entry in data: + if "webSocketDebuggerUrl" in entry: + ws_tail = entry["webSocketDebuggerUrl"].split("/devtools/")[-1] + if seed_key: + entry["webSocketDebuggerUrl"] = ( + f"{scheme}://{host}/fingerprint/{seed_key}/devtools/{ws_tail}" + ) + else: + entry["webSocketDebuggerUrl"] = f"{scheme}://{host}/devtools/{ws_tail}" + + return web.json_response(data) + + +# --------------------------------------------------------------------------- +# WebSocket proxy +# --------------------------------------------------------------------------- + +async def proxy_cdp_websocket( + client_ws: web.WebSocketResponse, + target_url: str, + label: str, +) -> None: + """Bidirectional WebSocket proxy between client and Chrome CDP.""" + import websockets + + try: + async with websockets.connect( + target_url, max_size=None, ping_interval=None, ping_timeout=None, + ) as cdp_ws: + logger.info("%s: connected to %s", label, target_url) + + async def client_to_cdp(): + try: + async for msg in client_ws: + if msg.type == aiohttp.WSMsgType.TEXT: + await cdp_ws.send(msg.data) + elif msg.type == aiohttp.WSMsgType.BINARY: + await cdp_ws.send(msg.data) + elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, aiohttp.WSMsgType.CLOSED): + break + except Exception as exc: + logger.debug("%s [c->cdp]: %s", label, exc) + + async def cdp_to_client(): + try: + async for msg in cdp_ws: + if isinstance(msg, str): + await client_ws.send_str(msg) + else: + await client_ws.send_bytes(msg) + except Exception as exc: + logger.debug("%s [cdp->c]: %s", label, exc) + + c2d = asyncio.create_task(client_to_cdp(), name="c2d") + d2c = asyncio.create_task(cdp_to_client(), name="d2c") + done, pending = await asyncio.wait( + [c2d, d2c], return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + logger.info("%s: disconnected", label) + + except Exception as exc: + logger.error("%s error: %s", label, exc) + + +async def handle_ws_default(request: web.Request) -> web.WebSocketResponse: + """WebSocket proxy for default (no-seed) Chrome: /devtools/{type}/{guid}""" + pool: ChromePool = request.app["pool"] + path = request.match_info.get("path", "") + + cp = await pool.get_or_launch(seed=None) + + ws = web.WebSocketResponse() + await ws.prepare(request) + + target_url = f"ws://127.0.0.1:{cp.cdp_port}/devtools/{path}" + await proxy_cdp_websocket(ws, target_url, f"CDP default [{path}]") + return ws + + +async def handle_ws_seed(request: web.Request) -> web.WebSocketResponse: + """WebSocket proxy for seed-specific Chrome: /fingerprint/{seed}/devtools/{type}/{guid}""" + pool: ChromePool = request.app["pool"] + seed = request.match_info["seed"] + path = request.match_info.get("path", "") + + cp = await pool.get_or_launch(seed=seed) + + ws = web.WebSocketResponse() + await ws.prepare(request) + + target_url = f"ws://127.0.0.1:{cp.cdp_port}/devtools/{path}" + await proxy_cdp_websocket(ws, target_url, f"CDP seed={seed} [{path}]") + return ws + + +async def on_shutdown(app: web.Application) -> None: + await app["pool"].shutdown() + + +# --------------------------------------------------------------------------- +# CLI arg parsing +# --------------------------------------------------------------------------- + +def parse_cli_args(argv: list[str]) -> tuple[dict, list[str]]: + """Parse cloakserve-specific args, return (config, passthrough_args).""" + config = { + "port": 9222, + "headless": True, + } + passthrough = [] + # Flags consumed by cloakserve (not passed to Chrome) + consumed_prefixes = ( + "--port=", + "--remote-debugging-port=", + "--remote-debugging-address=", + ) + + for arg in argv: + if arg.startswith("--port="): + config["port"] = int(arg.split("=", 1)[1]) + elif arg == "--headless=false" or arg == "--headless=False": + config["headless"] = False + passthrough.append(arg) + elif arg.startswith(consumed_prefixes): + pass # Strip these silently + else: + passthrough.append(arg) + + return config, passthrough + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main() -> None: + binary = ensure_binary() + config, global_args = parse_cli_args(sys.argv[1:]) + + pool = ChromePool( + binary=binary, + global_args=global_args, + headless=config["headless"], + ) + + app = web.Application() + app["pool"] = pool + app["port"] = config["port"] + + # Routes + app.router.add_get("/", handle_root) + app.router.add_get("/json/version", handle_json_version) + app.router.add_get("/json/version/", handle_json_version) + app.router.add_get("/json/list", handle_json_list) + app.router.add_get("/json/list/", handle_json_list) + app.router.add_get("/json", handle_json_list) + app.router.add_get("/json/", handle_json_list) + + # WebSocket routes — seed-specific (must be before default to match first) + app.router.add_get("/fingerprint/{seed}/devtools/{path:.+}", handle_ws_seed) + # WebSocket routes — default (no seed) + app.router.add_get("/devtools/{path:.+}", handle_ws_default) + + app.on_shutdown.append(on_shutdown) + + port = config["port"] + logger.info("CloakBrowser CDP multiplexer starting on port %d", port) + logger.info( + "Connect: playwright.chromium.connect_over_cdp(" + "\"http://localhost:%d?fingerprint=\")", + port, + ) + + web.run_app(app, host="0.0.0.0", port=port, print=None) + + +if __name__ == "__main__": + main() diff --git a/cloakbrowser/__init__.py b/cloakbrowser/__init__.py index 2fb96a2..d720486 100644 --- a/cloakbrowser/__init__.py +++ b/cloakbrowser/__init__.py @@ -11,7 +11,7 @@ Usage: browser.close() """ -from .browser import launch, launch_async, launch_context, launch_persistent_context, launch_persistent_context_async, ProxySettings +from .browser import launch, launch_async, launch_context, launch_persistent_context, launch_persistent_context_async, ProxySettings, build_args, maybe_resolve_geoip from .config import CHROMIUM_VERSION, get_default_stealth_args from .download import binary_info, check_for_update, clear_cache, ensure_binary from ._version import __version__ @@ -40,6 +40,8 @@ __all__ = [ "check_for_update", "CHROMIUM_VERSION", "get_default_stealth_args", + "build_args", + "maybe_resolve_geoip", "ProxySettings", "HumanConfig", "resolve_human_config", diff --git a/cloakbrowser/browser.py b/cloakbrowser/browser.py index 046c95b..c22a673 100644 --- a/cloakbrowser/browser.py +++ b/cloakbrowser/browser.py @@ -101,8 +101,8 @@ def launch( sync_playwright = _import_sync_playwright(_resolve_backend(backend)) binary_path = ensure_binary() - timezone, locale = _maybe_resolve_geoip(geoip, proxy, timezone, locale) - chrome_args = _build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) + timezone, locale = maybe_resolve_geoip(geoip, proxy, timezone, locale) + chrome_args = build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) logger.debug("Launching stealth Chromium (headless=%s, args=%d)", headless, len(chrome_args)) @@ -186,8 +186,8 @@ async def launch_async( # noqa: C901 async_playwright = _import_async_playwright(_resolve_backend(backend)) binary_path = ensure_binary() - timezone, locale = _maybe_resolve_geoip(geoip, proxy, timezone, locale) - chrome_args = _build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) + timezone, locale = maybe_resolve_geoip(geoip, proxy, timezone, locale) + chrome_args = build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) logger.debug("Launching stealth Chromium async (headless=%s, args=%d)", headless, len(chrome_args)) @@ -284,8 +284,8 @@ def launch_persistent_context( timezone = _resolve_timezone(timezone, kwargs) binary_path = ensure_binary() - timezone, locale = _maybe_resolve_geoip(geoip, proxy, timezone, locale) - chrome_args = _build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) + timezone, locale = maybe_resolve_geoip(geoip, proxy, timezone, locale) + chrome_args = build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) logger.debug( "Launching persistent stealth Chromium (headless=%s, user_data_dir=%s)", @@ -399,8 +399,8 @@ async def launch_persistent_context_async( timezone = _resolve_timezone(timezone, kwargs) binary_path = ensure_binary() - timezone, locale = _maybe_resolve_geoip(geoip, proxy, timezone, locale) - chrome_args = _build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) + timezone, locale = maybe_resolve_geoip(geoip, proxy, timezone, locale) + chrome_args = build_args(stealth_args, args, timezone=timezone, locale=locale, headless=headless) logger.debug( "Launching persistent stealth Chromium async (headless=%s, user_data_dir=%s)", @@ -497,7 +497,7 @@ def launch_context( # Resolve geoip BEFORE launch() to avoid double-resolution and ensure # resolved values flow to binary flags - timezone, locale = _maybe_resolve_geoip(geoip, proxy, timezone, locale) + timezone, locale = maybe_resolve_geoip(geoip, proxy, timezone, locale) # --fingerprint-timezone is process-wide (reads CommandLine in renderer), # so it applies to ALL contexts, not just the default one. # locale and timezone are set via binary flags only — no CDP emulation. @@ -590,7 +590,7 @@ def _ensure_proxy_scheme(proxy_url: str) -> str: return proxy_url if "://" in proxy_url else f"http://{proxy_url}" -def _maybe_resolve_geoip( +def maybe_resolve_geoip( geoip: bool, proxy: str | ProxySettings | None, timezone: str | None, @@ -614,7 +614,7 @@ def _maybe_resolve_geoip( return timezone, locale -def _build_args( +def build_args( stealth_args: bool, extra_args: list[str] | None, timezone: str | None = None, diff --git a/pyproject.toml b/pyproject.toml index fc0796e..efb2fa3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,7 @@ dependencies = [ [project.optional-dependencies] geoip = ["geoip2>=4.0"] patchright = ["patchright>=1.40"] +serve = ["aiohttp>=3.9", "websockets>=12.0"] dev = ["pytest>=7.0", "pytest-asyncio>=0.23"] [project.scripts] diff --git a/tests/test_build_args.py b/tests/test_build_args.py index d38ea7c..011dcdf 100644 --- a/tests/test_build_args.py +++ b/tests/test_build_args.py @@ -1,24 +1,24 @@ -"""Unit tests for _build_args timezone/locale injection and timezone alias.""" +"""Unit tests for build_args timezone/locale injection and timezone alias.""" -from cloakbrowser.browser import _build_args, _resolve_timezone +from cloakbrowser.browser import build_args, _resolve_timezone def test_timezone_injected(): """--fingerprint-timezone flag should appear when timezone is set.""" - args = _build_args(stealth_args=True, extra_args=None, timezone="America/New_York") + args = build_args(stealth_args=True, extra_args=None, timezone="America/New_York") assert "--fingerprint-timezone=America/New_York" in args def test_locale_injected(): """--lang and --fingerprint-locale flags should appear when locale is set.""" - args = _build_args(stealth_args=True, extra_args=None, locale="en-US") + args = build_args(stealth_args=True, extra_args=None, locale="en-US") assert "--lang=en-US" in args assert "--fingerprint-locale=en-US" in args def test_both_injected(): """Both flags should appear when both are set.""" - args = _build_args(stealth_args=True, extra_args=None, timezone="Europe/Berlin", locale="de-DE") + args = build_args(stealth_args=True, extra_args=None, timezone="Europe/Berlin", locale="de-DE") assert "--fingerprint-timezone=Europe/Berlin" in args assert "--lang=de-DE" in args assert "--fingerprint-locale=de-DE" in args @@ -26,7 +26,7 @@ def test_both_injected(): def test_timezone_independent_of_stealth_args(): """--fingerprint-timezone should be injected even when stealth_args=False.""" - args = _build_args(stealth_args=False, extra_args=None, timezone="America/New_York", locale="en-US") + args = build_args(stealth_args=False, extra_args=None, timezone="America/New_York", locale="en-US") assert "--fingerprint-timezone=America/New_York" in args assert "--lang=en-US" in args assert "--fingerprint-locale=en-US" in args @@ -36,7 +36,7 @@ def test_timezone_independent_of_stealth_args(): def test_no_flags_when_not_set(): """No timezone/lang/fingerprint-locale flags when params are None.""" - args = _build_args(stealth_args=True, extra_args=None) + args = build_args(stealth_args=True, extra_args=None) assert not any(a.startswith("--fingerprint-timezone=") for a in args) assert not any(a.startswith("--lang=") for a in args) assert not any(a.startswith("--fingerprint-locale=") for a in args) @@ -44,7 +44,7 @@ def test_no_flags_when_not_set(): def test_extra_args_preserved(): """Extra args should still be included alongside timezone/locale.""" - args = _build_args(stealth_args=True, extra_args=["--disable-gpu"], timezone="Asia/Tokyo", locale="ja-JP") + args = build_args(stealth_args=True, extra_args=["--disable-gpu"], timezone="Asia/Tokyo", locale="ja-JP") assert "--disable-gpu" in args assert "--fingerprint-timezone=Asia/Tokyo" in args assert "--lang=ja-JP" in args @@ -90,7 +90,7 @@ def test_resolve_both_none(): def test_user_fingerprint_overrides_default(): """User --fingerprint should override the random default seed.""" - args = _build_args(stealth_args=True, extra_args=["--fingerprint=99887"]) + args = build_args(stealth_args=True, extra_args=["--fingerprint=99887"]) fingerprint_args = [a for a in args if a.startswith("--fingerprint=")] assert len(fingerprint_args) == 1 assert fingerprint_args[0] == "--fingerprint=99887" @@ -98,7 +98,7 @@ def test_user_fingerprint_overrides_default(): def test_user_platform_overrides_default(): """User --fingerprint-platform should override the default.""" - args = _build_args(stealth_args=True, extra_args=["--fingerprint-platform=linux"]) + args = build_args(stealth_args=True, extra_args=["--fingerprint-platform=linux"]) platform_args = [a for a in args if a.startswith("--fingerprint-platform=")] assert len(platform_args) == 1 assert platform_args[0] == "--fingerprint-platform=linux" @@ -106,7 +106,7 @@ def test_user_platform_overrides_default(): def test_timezone_param_overrides_user_arg(): """Dedicated timezone param should override user arg.""" - args = _build_args( + args = build_args( stealth_args=True, extra_args=["--fingerprint-timezone=Europe/London"], timezone="America/New_York", @@ -118,7 +118,7 @@ def test_timezone_param_overrides_user_arg(): def test_locale_param_overrides_user_arg(): """Dedicated locale param should override user --lang and --fingerprint-locale args.""" - args = _build_args( + args = build_args( stealth_args=True, extra_args=["--lang=de-DE", "--fingerprint-locale=de-DE"], locale="en-US", @@ -133,7 +133,7 @@ def test_locale_param_overrides_user_arg(): def test_no_duplicate_flags(): """No flag key should appear more than once in the output.""" - args = _build_args( + args = build_args( stealth_args=True, extra_args=["--fingerprint=99887", "--fingerprint-timezone=UTC", "--lang=fr-FR"], timezone="Europe/Berlin", @@ -145,7 +145,7 @@ def test_no_duplicate_flags(): def test_non_value_flags_preserved(): """Flags without = should be preserved without dedup issues.""" - args = _build_args(stealth_args=True, extra_args=["--disable-gpu", "--no-zygote"]) + args = build_args(stealth_args=True, extra_args=["--disable-gpu", "--no-zygote"]) assert "--disable-gpu" in args assert "--no-zygote" in args assert "--no-sandbox" in args @@ -156,5 +156,5 @@ def test_override_logs_debug(caplog): import logging with caplog.at_level(logging.DEBUG, logger="cloakbrowser"): - _build_args(stealth_args=True, extra_args=["--fingerprint=99887"]) + build_args(stealth_args=True, extra_args=["--fingerprint=99887"]) assert any("--fingerprint=" in r.message and "99887" in r.message for r in caplog.records) diff --git a/tests/test_cloakserve.py b/tests/test_cloakserve.py new file mode 100644 index 0000000..c857442 --- /dev/null +++ b/tests/test_cloakserve.py @@ -0,0 +1,171 @@ +"""Unit tests for cloakserve — parse_connection_params, parse_cli_args, URL rewriting.""" + +import importlib.machinery +import importlib.util +import sys +from pathlib import Path + +import pytest + +aiohttp = pytest.importorskip("aiohttp", reason="cloakserve requires aiohttp (install with .[serve])") + +# Load cloakserve as a module from bin/ (no .py extension). +_bin_path = str(Path(__file__).resolve().parents[1] / "bin" / "cloakserve") +_loader = importlib.machinery.SourceFileLoader("cloakserve", _bin_path) +_spec = importlib.util.spec_from_file_location("cloakserve", _bin_path, loader=_loader) +_mod = importlib.util.module_from_spec(_spec) +sys.modules["cloakserve"] = _mod +_loader.exec_module(_mod) + +parse_connection_params = _mod.parse_connection_params +parse_cli_args = _mod.parse_cli_args + + +# --------------------------------------------------------------------------- +# parse_connection_params +# --------------------------------------------------------------------------- + + +class TestParseConnectionParams: + def test_empty_query(self): + result = parse_connection_params("") + assert result["seed"] is None + assert result["extra_args"] == [] + + def test_fingerprint_seed(self): + result = parse_connection_params("fingerprint=12345") + assert result["seed"] == "12345" + + def test_timezone_and_locale(self): + result = parse_connection_params("fingerprint=1&timezone=Asia/Tokyo&locale=ja-JP") + assert result["timezone"] == "Asia/Tokyo" + assert result["locale"] == "ja-JP" + + def test_proxy(self): + result = parse_connection_params("proxy=http://proxy:8080") + assert result["proxy"] == "http://proxy:8080" + + def test_geoip_true_variants(self): + for val in ("true", "1", "yes", "True", "YES"): + result = parse_connection_params(f"geoip={val}") + assert result["geoip"] is True, f"geoip={val} should be True" + + def test_geoip_false(self): + for val in ("false", "0", "no", "anything"): + result = parse_connection_params(f"geoip={val}") + assert result["geoip"] is False, f"geoip={val} should be False" + + def test_generic_fingerprint_params(self): + qs = "fingerprint=1&platform=windows&hardware-concurrency=8&gpu-vendor=NVIDIA" + result = parse_connection_params(qs) + assert "--fingerprint-platform=windows" in result["extra_args"] + assert "--fingerprint-hardware-concurrency=8" in result["extra_args"] + assert "--fingerprint-gpu-vendor=NVIDIA" in result["extra_args"] + + def test_special_params_not_in_extra_args(self): + qs = "fingerprint=1&timezone=UTC&locale=en-US&proxy=http://x:1&geoip=true" + result = parse_connection_params(qs) + assert result["extra_args"] == [] + + def test_multiple_values_takes_first(self): + result = parse_connection_params("fingerprint=111&fingerprint=222") + assert result["seed"] == "111" + + +# --------------------------------------------------------------------------- +# parse_cli_args +# --------------------------------------------------------------------------- + + +class TestParseCliArgs: + def test_defaults(self): + config, passthrough = parse_cli_args([]) + assert config["port"] == 9222 + assert config["headless"] is True + assert passthrough == [] + + def test_custom_port(self): + config, _ = parse_cli_args(["--port=8080"]) + assert config["port"] == 8080 + + def test_headless_false(self): + config, passthrough = parse_cli_args(["--headless=false"]) + assert config["headless"] is False + # headless flag still passed through to Chrome + assert "--headless=false" in passthrough + + def test_strips_remote_debugging_flags(self): + args = ["--remote-debugging-port=9999", "--remote-debugging-address=0.0.0.0", "--no-sandbox"] + config, passthrough = parse_cli_args(args) + assert passthrough == ["--no-sandbox"] + + def test_passthrough_args(self): + args = ["--no-sandbox", "--disable-gpu", "--fingerprint=999"] + _, passthrough = parse_cli_args(args) + assert passthrough == args + + def test_port_not_in_passthrough(self): + _, passthrough = parse_cli_args(["--port=9222", "--no-sandbox"]) + assert "--port=9222" not in passthrough + assert "--no-sandbox" in passthrough + + +# --------------------------------------------------------------------------- +# URL rewriting logic (pure string manipulation, extracted from handlers) +# --------------------------------------------------------------------------- + + +class TestURLRewriting: + """Test the URL rewriting logic used by /json/version and /json/list.""" + + def _rewrite_version(self, orig_ws: str, host: str, seed: str | None, scheme: str = "ws") -> str: + """Replicate the URL rewrite logic from handle_json_version.""" + if seed: + ws_path = f"fingerprint/{seed}/devtools/browser" + else: + ws_path = "devtools/browser" + guid = orig_ws.rsplit("/", 1)[-1] if "/devtools/" in orig_ws else "" + return f"{scheme}://{host}/{ws_path}/{guid}" + + def _rewrite_list_entry(self, orig_ws: str, host: str, seed: str | None, scheme: str = "ws") -> str: + """Replicate the URL rewrite logic from handle_json_list.""" + ws_tail = orig_ws.split("/devtools/")[-1] + if seed: + return f"{scheme}://{host}/fingerprint/{seed}/devtools/{ws_tail}" + else: + return f"{scheme}://{host}/devtools/{ws_tail}" + + def test_version_rewrite_with_seed(self): + orig = "ws://127.0.0.1:5100/devtools/browser/abc-123" + result = self._rewrite_version(orig, "container:9222", "12345") + assert result == "ws://container:9222/fingerprint/12345/devtools/browser/abc-123" + + def test_version_rewrite_no_seed(self): + orig = "ws://127.0.0.1:5100/devtools/browser/abc-123" + result = self._rewrite_version(orig, "container:9222", None) + assert result == "ws://container:9222/devtools/browser/abc-123" + + def test_list_rewrite_page_with_seed(self): + orig = "ws://127.0.0.1:5100/devtools/page/DEF-456" + result = self._rewrite_list_entry(orig, "host:9222", "99") + assert result == "ws://host:9222/fingerprint/99/devtools/page/DEF-456" + + def test_list_rewrite_page_no_seed(self): + orig = "ws://127.0.0.1:5100/devtools/page/DEF-456" + result = self._rewrite_list_entry(orig, "host:9222", None) + assert result == "ws://host:9222/devtools/page/DEF-456" + + def test_list_rewrite_browser(self): + orig = "ws://127.0.0.1:5100/devtools/browser/XYZ" + result = self._rewrite_list_entry(orig, "host:9222", "seed1") + assert result == "ws://host:9222/fingerprint/seed1/devtools/browser/XYZ" + + def test_wss_scheme_version(self): + orig = "ws://127.0.0.1:5100/devtools/browser/abc-123" + result = self._rewrite_version(orig, "host:443", "seed1", scheme="wss") + assert result == "wss://host:443/fingerprint/seed1/devtools/browser/abc-123" + + def test_wss_scheme_list(self): + orig = "ws://127.0.0.1:5100/devtools/page/DEF-456" + result = self._rewrite_list_entry(orig, "host:443", "seed1", scheme="wss") + assert result == "wss://host:443/fingerprint/seed1/devtools/page/DEF-456" diff --git a/tests/test_geoip.py b/tests/test_geoip.py index 179613b..bd92d61 100644 --- a/tests/test_geoip.py +++ b/tests/test_geoip.py @@ -4,7 +4,7 @@ from unittest.mock import patch import pytest -from cloakbrowser.browser import _maybe_resolve_geoip +from cloakbrowser.browser import maybe_resolve_geoip from cloakbrowser.geoip import ( COUNTRY_LOCALE_MAP, _is_private_ip, @@ -92,25 +92,25 @@ def test_resolve_geo_returns_none_when_db_missing(): # --------------------------------------------------------------------------- -# _maybe_resolve_geoip (browser.py helper) +# maybe_resolve_geoip (browser.py helper) # --------------------------------------------------------------------------- def test_maybe_resolve_skips_when_geoip_false(): - tz, loc = _maybe_resolve_geoip(False, "http://proxy:8080", None, None) + tz, loc = maybe_resolve_geoip(False, "http://proxy:8080", None, None) assert tz is None assert loc is None def test_maybe_resolve_skips_when_no_proxy(): - tz, loc = _maybe_resolve_geoip(True, None, None, None) + tz, loc = maybe_resolve_geoip(True, None, None, None) assert tz is None assert loc is None def test_maybe_resolve_skips_when_both_explicit(): """Explicit values should not trigger geoip resolution.""" - tz, loc = _maybe_resolve_geoip(True, "http://proxy:8080", "Europe/Berlin", "de-DE") + tz, loc = maybe_resolve_geoip(True, "http://proxy:8080", "Europe/Berlin", "de-DE") assert tz == "Europe/Berlin" assert loc == "de-DE" @@ -118,7 +118,7 @@ def test_maybe_resolve_skips_when_both_explicit(): def test_maybe_resolve_fills_missing_timezone(): """When only locale is explicit, geoip should fill timezone.""" with patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("America/New_York", "en-US")): - tz, loc = _maybe_resolve_geoip(True, "http://proxy:8080", None, "fr-FR") + tz, loc = maybe_resolve_geoip(True, "http://proxy:8080", None, "fr-FR") assert tz == "America/New_York" assert loc == "fr-FR" # Explicit wins @@ -126,7 +126,7 @@ def test_maybe_resolve_fills_missing_timezone(): def test_maybe_resolve_fills_missing_locale(): """When only timezone is explicit, geoip should fill locale.""" with patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("America/New_York", "en-US")): - tz, loc = _maybe_resolve_geoip(True, "http://proxy:8080", "Asia/Tokyo", None) + tz, loc = maybe_resolve_geoip(True, "http://proxy:8080", "Asia/Tokyo", None) assert tz == "Asia/Tokyo" # Explicit wins assert loc == "en-US" @@ -134,7 +134,7 @@ def test_maybe_resolve_fills_missing_locale(): def test_maybe_resolve_fills_both(): """When neither is set, geoip should fill both.""" with patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("Europe/Berlin", "de-DE")): - tz, loc = _maybe_resolve_geoip(True, "http://proxy:8080", None, None) + tz, loc = maybe_resolve_geoip(True, "http://proxy:8080", None, None) assert tz == "Europe/Berlin" assert loc == "de-DE" diff --git a/tests/test_launch_context.py b/tests/test_launch_context.py index 8c80f22..f930612 100644 --- a/tests/test_launch_context.py +++ b/tests/test_launch_context.py @@ -114,7 +114,7 @@ def test_color_scheme(mock_launch, _mock_bin): assert ctx_kwargs[1]["color_scheme"] == "dark" -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=("Europe/Berlin", "de-DE")) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=("Europe/Berlin", "de-DE")) @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") @patch("cloakbrowser.browser.launch") def test_geoip_resolution(mock_launch, _mock_bin, _mock_geoip): diff --git a/tests/test_persistent_context.py b/tests/test_persistent_context.py index cb58c22..42944dc 100644 --- a/tests/test_persistent_context.py +++ b/tests/test_persistent_context.py @@ -26,7 +26,7 @@ def _make_mock_pw_and_context(): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_args_built(_mock_geoip, _mock_bin): """Stealth args + extra args combined correctly.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -42,7 +42,7 @@ def test_persistent_context_args_built(_mock_geoip, _mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_default_viewport(_mock_geoip, _mock_bin): """DEFAULT_VIEWPORT applied when no viewport given.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -56,7 +56,7 @@ def test_persistent_context_default_viewport(_mock_geoip, _mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_custom_viewport(_mock_geoip, _mock_bin): """Custom viewport overrides DEFAULT_VIEWPORT.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -71,7 +71,7 @@ def test_persistent_context_custom_viewport(_mock_geoip, _mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_user_agent(_mock_geoip, _mock_bin): """user_agent forwarded to launch_persistent_context().""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -103,7 +103,7 @@ def test_persistent_context_locale_and_timezone(_mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_color_scheme(_mock_geoip, _mock_bin): """color_scheme forwarded correctly.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -116,7 +116,7 @@ def test_persistent_context_color_scheme(_mock_geoip, _mock_bin): assert call_kwargs["color_scheme"] == "dark" -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=("Europe/Berlin", "de-DE")) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=("Europe/Berlin", "de-DE")) @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") def test_persistent_context_geoip(_mock_bin, _mock_geoip): """geoip fills missing tz/locale — flows to binary args, not CDP context.""" @@ -150,7 +150,7 @@ def test_persistent_context_timezone_id_alias(_mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_close_stops_pw(_mock_geoip, _mock_bin): """context.close() also calls pw.stop().""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -166,7 +166,7 @@ def test_persistent_context_close_stops_pw(_mock_geoip, _mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_proxy_string(_mock_geoip, _mock_bin): """Proxy string parsed and passed.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -182,7 +182,7 @@ def test_persistent_context_proxy_string(_mock_geoip, _mock_bin): @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) def test_persistent_context_proxy_dict(_mock_geoip, _mock_bin): """Proxy dict passed through.""" pw_cm, pw, context = _make_mock_pw_and_context() @@ -213,7 +213,7 @@ def _make_mock_async_pw_and_context(): @pytest.mark.asyncio @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) async def test_persistent_context_async_args_built(_mock_geoip, _mock_bin): """Async launch builds args correctly.""" pw_cm, pw, context = _make_mock_async_pw_and_context() @@ -229,7 +229,7 @@ async def test_persistent_context_async_args_built(_mock_geoip, _mock_bin): @pytest.mark.asyncio @patch("cloakbrowser.browser.ensure_binary", return_value="/fake/chrome") -@patch("cloakbrowser.browser._maybe_resolve_geoip", return_value=(None, None)) +@patch("cloakbrowser.browser.maybe_resolve_geoip", return_value=(None, None)) async def test_persistent_context_async_close_stops_pw(_mock_geoip, _mock_bin): """await context.close() calls await pw.stop().""" pw_cm, pw, context = _make_mock_async_pw_and_context() diff --git a/tests/test_proxy.py b/tests/test_proxy.py index 17d9a7b..1275e1f 100644 --- a/tests/test_proxy.py +++ b/tests/test_proxy.py @@ -2,7 +2,7 @@ from unittest.mock import patch -from cloakbrowser.browser import _build_proxy_kwargs, _maybe_resolve_geoip, _parse_proxy_url +from cloakbrowser.browser import _build_proxy_kwargs, maybe_resolve_geoip, _parse_proxy_url class TestParseProxyUrl: @@ -70,7 +70,7 @@ class TestBuildProxyKwargs: class TestMaybeResolveGeoip: @patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("America/New_York", "en-US")) def test_geoip_with_string_proxy(self, mock_geo): - tz, locale = _maybe_resolve_geoip(True, "http://proxy:8080", None, None) + tz, locale = maybe_resolve_geoip(True, "http://proxy:8080", None, None) mock_geo.assert_called_once_with("http://proxy:8080") assert tz == "America/New_York" assert locale == "en-US" @@ -78,31 +78,31 @@ class TestMaybeResolveGeoip: @patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("Europe/London", "en-GB")) def test_geoip_with_dict_proxy_extracts_server(self, mock_geo): proxy_dict = {"server": "http://proxy:8080", "bypass": ".google.com"} - tz, locale = _maybe_resolve_geoip(True, proxy_dict, None, None) + tz, locale = maybe_resolve_geoip(True, proxy_dict, None, None) mock_geo.assert_called_once_with("http://proxy:8080") assert tz == "Europe/London" assert locale == "en-GB" def test_geoip_disabled_skips_resolution(self): - tz, locale = _maybe_resolve_geoip(False, "http://proxy:8080", None, None) + tz, locale = maybe_resolve_geoip(False, "http://proxy:8080", None, None) assert tz is None assert locale is None def test_geoip_no_proxy_skips_resolution(self): - tz, locale = _maybe_resolve_geoip(True, None, None, None) + tz, locale = maybe_resolve_geoip(True, None, None, None) assert tz is None assert locale is None @patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("Asia/Tokyo", "ja-JP")) def test_geoip_preserves_explicit_timezone(self, mock_geo): - tz, locale = _maybe_resolve_geoip(True, "http://proxy:8080", "Europe/Berlin", None) + tz, locale = maybe_resolve_geoip(True, "http://proxy:8080", "Europe/Berlin", None) assert tz == "Europe/Berlin" assert locale == "ja-JP" @patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("America/New_York", "en-US")) def test_geoip_normalizes_bare_proxy_with_creds(self, mock_geo): # "user:pass@host:port" must be normalized to http:// before geoip lookup. - tz, locale = _maybe_resolve_geoip(True, "user:pass@proxy:8080", None, None) + tz, locale = maybe_resolve_geoip(True, "user:pass@proxy:8080", None, None) mock_geo.assert_called_once_with("http://user:pass@proxy:8080") assert tz == "America/New_York" assert locale == "en-US" @@ -110,7 +110,7 @@ class TestMaybeResolveGeoip: @patch("cloakbrowser.geoip.resolve_proxy_geo", return_value=("America/New_York", "en-US")) def test_geoip_normalizes_schemeless_proxy_no_creds(self, mock_geo): # "host:port" (no @ and no scheme) must also be normalized. - tz, locale = _maybe_resolve_geoip(True, "proxy:8080", None, None) + tz, locale = maybe_resolve_geoip(True, "proxy:8080", None, None) mock_geo.assert_called_once_with("http://proxy:8080") assert tz == "America/New_York"