camoufox/pythonlib/camoufox/sync_api.py

import json as _json
import urllib.request
from typing import Any, Dict, List, Optional, Union, overload
from urllib.parse import urlparse

from playwright.sync_api import (
    Browser,
    BrowserContext,
    Playwright,
    PlaywrightContextManager,
)
from typing_extensions import Literal

from camoufox.virtdisplay import VirtualDisplay

from .fingerprints import generate_context_fingerprint
from .utils import (
    attach_no_viewport_default,
    launch_options,
    spoofs_window_dimensions,
    sync_attach_vd,
)


class Camoufox(PlaywrightContextManager):
    """
    Wrapper around playwright.sync_api.PlaywrightContextManager that automatically
    launches a browser and closes it when the context manager is exited.
    """

    def __init__(self, **launch_options):
        super().__init__()
        self.launch_options = launch_options
        self.browser: Optional[Union[Browser, BrowserContext]] = None

    def __enter__(self) -> Union[Browser, BrowserContext]:
        super().__enter__()
        try:
            self.browser = NewBrowser(self._playwright, **self.launch_options)
        except BaseException as e:
            # Any launch failure (InvalidProxy, missing browser, bad options, ...)
            # must tear down the playwright session started above. Leaking it leaves
            # the sync API's event loop in a "running" state, so every later sync
            # Camoufox/Playwright start in this thread fails with "Sync API inside
            # the asyncio loop" until the process restarts (#82).
            super().__exit__(type(e), e, e.__traceback__)
            raise
        return self.browser

    def __exit__(self, *args: Any):
        # Run the base teardown even if browser.close() raises (e.g. the browser
        # process already crashed). Skipping it leaks the sync API's event loop in a
        # "running" state, so every later sync Camoufox/Playwright start in the same
        # thread fails with "Sync API inside the asyncio loop" until process restart.
        try:
            if self.browser:
                self.browser.close()
        finally:
            super().__exit__(*args)


@overload
def NewBrowser(
    playwright: Playwright,
    *,
    from_options: Optional[Dict[str, Any]] = None,
    persistent_context: Literal[False] = False,
    **kwargs,
) -> Browser: ...


@overload
def NewBrowser(
    playwright: Playwright,
    *,
    from_options: Optional[Dict[str, Any]] = None,
    persistent_context: Literal[True],
    **kwargs,
) -> BrowserContext: ...


def NewBrowser(
    playwright: Playwright,
    *,
    headless: Optional[Union[bool, Literal['virtual']]] = None,
    from_options: Optional[Dict[str, Any]] = None,
    persistent_context: bool = False,
    debug: Optional[bool] = None,
    **kwargs,
) -> Union[Browser, BrowserContext]:
    """
    Launches a new browser instance for Camoufox given a set of launch options.

    Parameters:
        from_options (Dict[str, Any]):
            A set of launch options generated by `launch_options()` to use
        persistent_context (bool):
            Whether to use a persistent context.
        **kwargs:
            All other keyword arugments passed to `launch_options()`.
    """
    if headless == 'virtual':
        virtual_display = VirtualDisplay(debug=debug)
        kwargs['virtual_display'] = virtual_display.get()
        headless = False
    else:
        virtual_display = None

    if not from_options:
        from_options = launch_options(headless=headless, debug=debug, **kwargs)

    # Playwright's default viewport deadlocks Juggler when the window is spoofed
    # to a different size (daijro/camoufox#666), so default to no_viewport.
    no_viewport_default = spoofs_window_dimensions(from_options)

    # Persistent context
    if persistent_context:
        if no_viewport_default and not ('viewport' in from_options or 'no_viewport' in from_options):
            from_options = {**from_options, 'no_viewport': True}
        context = playwright.firefox.launch_persistent_context(**from_options)
        return sync_attach_vd(context, virtual_display)

    # Browser
    browser = playwright.firefox.launch(**from_options)
    if no_viewport_default:
        attach_no_viewport_default(browser)
    return sync_attach_vd(browser, virtual_display)


def _proxy_url_with_creds(proxy: Dict[str, str]) -> str:
    """Builds a proxy URL string with embedded credentials."""
    parsed = urlparse(proxy.get("server", ""))
    user = proxy.get("username", "")
    pwd = proxy.get("password", "")
    if user and pwd:
        return f"{parsed.scheme}://{user}:{pwd}@{parsed.netloc}"
    return proxy.get("server", "")


def _resolve_proxy_geo(proxy: Dict[str, str]) -> Dict[str, Optional[str]]:
    """Queries ip-api.com through the proxy for the exit IP and timezone."""
    proxy_url = _proxy_url_with_creds(proxy)
    handler = urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
    opener = urllib.request.build_opener(handler)
    try:
        with opener.open("http://ip-api.com/json?fields=query,timezone", timeout=10) as resp:
            data = _json.loads(resp.read())
            return {"ip": data.get("query") or None, "timezone": data.get("timezone") or None}
    except Exception:
        return {"ip": None, "timezone": None}


def NewContext(
    browser: Browser,
    *,
    preset: Optional[Dict[str, Any]] = None,
    os: Optional[str] = None,
    ff_version: Optional[str] = None,
    webrtc_ip: Optional[str] = None,
    proxy: Optional[Dict[str, str]] = None,
    geolocation: Optional[Dict[str, float]] = None,
    **context_kwargs: Any,
) -> BrowserContext:
    """
    Creates a new browser context with a unique fingerprint identity.

    Each context gets its own real fingerprint preset
    with unique seeds for audio, canvas, and font spacing noise. All values are applied
    via addInitScript so they self-destruct before page scripts can detect them.

    Parameters:
        browser: A Browser instance from NewBrowser or Camoufox.
        preset: A specific fingerprint preset dict to use. If None, picks randomly.
        os: Target OS for preset selection ("windows", "macos", "linux").
        ff_version: Firefox version string for UA patching.
        webrtc_ip: IPv4 address to spoof for WebRTC ICE candidates.
        proxy: Per-context proxy (Playwright format: {"server": "...", "username": "...", "password": "..."}).
        geolocation: Per-context geolocation ({"latitude": float, "longitude": float}).
        **context_kwargs: Additional Playwright new_context() options.
    """
    # Auto-derive WebRTC IP and timezone from proxy's exit IP when not explicitly provided
    if proxy and (not webrtc_ip or "timezone_id" not in context_kwargs):
        geo = _resolve_proxy_geo(proxy)
        if not webrtc_ip:
            webrtc_ip = geo["ip"]
        if "timezone_id" not in context_kwargs and geo["timezone"]:
            context_kwargs["timezone_id"] = geo["timezone"]

    fp = generate_context_fingerprint(preset=preset, os=os, ff_version=ff_version, webrtc_ip=webrtc_ip)

    # Merge generated context options with user overrides (user wins)
    opts: Dict[str, Any] = {**fp['context_options'], **context_kwargs}
    if proxy:
        opts['proxy'] = proxy
    if geolocation:
        opts['geolocation'] = geolocation
        opts.setdefault('permissions', ['geolocation'])

    context = browser.new_context(**opts)
    context.add_init_script(fp['init_script'])
    return context