camoufox/pythonlib/camoufox/sync_api.py

from typing import Any, Dict, Optional, Tuple, Union, overload

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 .ip import Proxy, proxy_exit_geo
from .utils import (
    attach_no_viewport_default,
    attach_stock_media_defaults,
    launch_options,
    STOCK_MEDIA_DEFAULTS,
    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:
        # Opt-in (2026-09-17). Pinning keeps the identity's core count by
        # constraining the browser to that many cores; it costs real CPU, needs
        # a launch lock, and does nothing on macOS. What it defends against is a
        # page timing N parallel workers, which is expensive and noisy on a busy
        # machine. Off, the host's own snapped count is reported, so reported
        # and measurable still agree -- the identity just loses that one draw.
        kwargs.setdefault('pin_cpu_cores', False)
        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)

    # Pin the driver (and so the browser it is about to spawn) to as many
    # cores as the identity reports, so measurable parallelism matches
    # navigator.hardwareConcurrency; the driver gets its cores back afterwards.
    from . import cpu_affinity
    from .utils import driver_pid, pinned_core_count

    pin_to = pinned_core_count(from_options)
    pid = driver_pid(playwright) if pin_to else None
    previous = cpu_affinity.pin(pid, pin_to) if pid else None
    try:
        # 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}
            # The persistent context is created by the launch itself, so its media
            # features come from these options rather than from new_context().
            from_options = {
                **{k: v for k, v in STOCK_MEDIA_DEFAULTS.items() if k not in from_options},
                **from_options,
            }
            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)
        attach_stock_media_defaults(browser)
        return sync_attach_vd(browser, virtual_display)
    finally:
        if pid:
            cpu_affinity.restore(pid, previous)


def _resolve_proxy_geo(proxy: Dict[str, str]) -> Tuple[str, str]:
    """The proxy's exit IP and timezone."""
    return proxy_exit_geo(Proxy(**proxy).as_string())


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 its own audio noise seed. 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": "..."}).
            Unless webrtc_ip and timezone_id are both given, they are looked up from the
            proxy's exit IP; InvalidIP is raised if that lookup fails.
        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):
        exit_ip, timezone = _resolve_proxy_geo(proxy)
        webrtc_ip = webrtc_ip or exit_ip
        context_kwargs.setdefault("timezone_id", 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