mirror of
https://github.com/CloakHQ/CloakBrowser.git
synced 2026-10-22 00:22:17 -05:00
CloakHQ
89fb671af8
Python async launchers ran the binary/license check and the GeoIP and WebRTC exit-IP lookups on the caller's event loop, freezing every other coroutine for up to 20s per launch. Run them via asyncio.to_thread. .NET: license validation and Pro version lookups inside async download methods now run via Task.Run instead of holding the caller's thread.
1862 lines
76 KiB
Python
1862 lines
76 KiB
Python
"""Core browser launch functions for cloakbrowser.
|
|
|
|
Provides launch() and launch_async() — thin wrappers around Playwright
|
|
that use our patched stealth Chromium binary instead of stock Chromium.
|
|
|
|
Usage:
|
|
from cloakbrowser import launch
|
|
|
|
browser = launch()
|
|
page = browser.new_page()
|
|
page.goto("https://protected-site.com")
|
|
browser.close()
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import logging
|
|
import os
|
|
import sys
|
|
from typing import Any, Literal, TypedDict
|
|
from urllib.parse import quote, unquote, urlparse, urlunparse
|
|
|
|
from .config import (
|
|
DEFAULT_VIEWPORT,
|
|
IGNORE_DEFAULT_ARGS,
|
|
binary_supports_headless_no_viewport,
|
|
binary_supports_http_proxy_inline_auth,
|
|
binary_supports_maximized_window,
|
|
get_default_stealth_args,
|
|
persistent_seed_args,
|
|
)
|
|
from .download import ensure_binary
|
|
from .license import (
|
|
CloakBrowserLicenseError,
|
|
build_launch_env,
|
|
license_error_for_code,
|
|
license_error_message,
|
|
mint_denial_file,
|
|
read_denial_file,
|
|
resolve_license_key,
|
|
)
|
|
from .human.config import HumanConfigOverrides, HumanPreset
|
|
from .widevine import seed_widevine_hint
|
|
|
|
logger = logging.getLogger("cloakbrowser")
|
|
|
|
|
|
def _license_error(exc: BaseException) -> CloakBrowserLicenseError | None:
|
|
"""Return a CloakBrowserLicenseError if a launch failure was a license deny.
|
|
|
|
The Pro binary exits with a distinct code (76-79) on a license problem;
|
|
Playwright embeds that code in the launch-failure text. Returns None for any
|
|
other failure so a genuine crash propagates unchanged.
|
|
"""
|
|
msg = license_error_message(str(exc))
|
|
return CloakBrowserLicenseError(msg) if msg is not None else None
|
|
|
|
|
|
# Factory methods whose return value is itself a handle the user drives (a page
|
|
# or context handed back after launch). Guarding these *deeply* — guarding the
|
|
# object they return — is what lets a denial that lands AFTER the handshake
|
|
# surface on the returned page's first call, not only on a second new_page().
|
|
_GUARD_FACTORY_METHODS = ("new_page", "new_context")
|
|
|
|
# Never wrap these. ``close`` already carries teardown logic and must not be
|
|
# turned into a license error mid-cleanup; the event-listener surface is called
|
|
# internally by Playwright (event dispatch), so throwing a license error from
|
|
# inside it would crash the driver rather than surface cleanly to the user.
|
|
_GUARD_SKIP_METHODS = frozenset({
|
|
"close", "on", "once", "remove_listener",
|
|
})
|
|
|
|
|
|
class _Guarded:
|
|
"""A non-descriptor callable that wraps the guard closure.
|
|
|
|
The guard is stored back on the target as a plain attribute, and humanize
|
|
copies it into a ``type("Originals", ...)()`` holder and reads it off that
|
|
*instance*. Anything with ``__get__`` (a bare function, or ``functools.partial``
|
|
on Python 3.14+, where partial became a method descriptor) would re-bind to the
|
|
holder and inject a spurious first positional arg. This class has no ``__get__``,
|
|
so it stays inert wherever it is stored — on any Python version. Works for both
|
|
the sync guard and the async guard: for the async case ``__call__`` returns the
|
|
coroutine the closure produces, which the caller awaits. (issue #488)
|
|
"""
|
|
|
|
__slots__ = ("_fn",)
|
|
|
|
def __init__(self, fn: Any) -> None:
|
|
self._fn = fn
|
|
|
|
def __call__(self, *args: Any, **kwargs: Any) -> Any:
|
|
return self._fn(*args, **kwargs)
|
|
|
|
|
|
def _denial_license_error(denial_path: str) -> CloakBrowserLicenseError | None:
|
|
"""Read the denial file and map it to a license error, or None."""
|
|
code = read_denial_file(denial_path)
|
|
return license_error_for_code(code) if code is not None else None
|
|
|
|
|
|
def _guardable_method_names(target: Any) -> list[str]:
|
|
"""Public, non-property callables on *target* — the methods that can raise a
|
|
TargetClosedError once the browser dies. Properties are skipped by name so
|
|
their getters are never triggered here; this covers every real method
|
|
without enumerating them by hand.
|
|
"""
|
|
names = []
|
|
for name in dir(target):
|
|
if name.startswith("_") or name in _GUARD_SKIP_METHODS:
|
|
continue
|
|
if isinstance(getattr(type(target), name, None), property):
|
|
continue
|
|
try:
|
|
attr = getattr(target, name)
|
|
except Exception:
|
|
continue
|
|
if callable(attr) and not isinstance(attr, type):
|
|
names.append(name)
|
|
return names
|
|
|
|
|
|
def _install_license_guard(target: Any, denial_path: str) -> None:
|
|
"""Guard every public method of *target* (a browser, context, or page) so a
|
|
post-handshake license denial surfaces as CloakBrowserLicenseError on the
|
|
user's first failing call — whichever method that is.
|
|
|
|
A denial that lands after Playwright connected kills the browser without a
|
|
launch failure; the user's next call would otherwise raise a bare
|
|
TargetClosedError. Each wrapped method, on any exception, checks the denial
|
|
file the binary wrote and re-raises as CloakBrowserLicenseError when a
|
|
license code is present — otherwise the original error propagates unchanged,
|
|
so a genuine crash is never mislabelled. Factory methods (new_page/
|
|
new_context) additionally guard the object they return, so a page or context
|
|
obtained after launch is covered too.
|
|
"""
|
|
for name in _guardable_method_names(target):
|
|
original = getattr(target, name)
|
|
deep = name in _GUARD_FACTORY_METHODS
|
|
|
|
def make(original: Any, deep: bool) -> Any:
|
|
def guarded(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
result = original(*args, **kwargs)
|
|
except Exception as exc:
|
|
lic = _denial_license_error(denial_path)
|
|
if lic is not None:
|
|
raise lic from exc
|
|
raise
|
|
# Even on success the browser may already be denied: the binary
|
|
# writes the denial file the instant it's over cap but keeps
|
|
# serving (blank) responses for ~1s before it exits. Check after
|
|
# every call so the denial surfaces on the first call that runs
|
|
# once the file exists, not only after the process has died.
|
|
lic = _denial_license_error(denial_path)
|
|
if lic is not None:
|
|
raise lic
|
|
if deep and result is not None:
|
|
_install_license_guard(result, denial_path)
|
|
return result
|
|
return guarded
|
|
|
|
# Wrap so the guard is NOT a descriptor. Playwright's real methods are
|
|
# bound methods; humanize copies them into a class attribute
|
|
# (type("Originals", ...)) and reads them back off an instance. A bare
|
|
# function — or a functools.partial on Python 3.14+, where partial became
|
|
# a method descriptor — would re-bind there and inject a spurious first
|
|
# positional arg. _Guarded has no __get__, so it stays inert. (issue #488)
|
|
setattr(target, name, _Guarded(make(original, deep)))
|
|
|
|
|
|
def _install_license_guard_async(target: Any, denial_path: str) -> None:
|
|
"""Async variant of :func:`_install_license_guard`. Only coroutine methods
|
|
are wrapped — sync helpers (``is_closed``, ``on`` …) never do a CDP round
|
|
trip and can't raise the disconnect error, and awaiting them would break.
|
|
"""
|
|
for name in _guardable_method_names(target):
|
|
original = getattr(target, name)
|
|
if not inspect.iscoroutinefunction(original):
|
|
continue
|
|
deep = name in _GUARD_FACTORY_METHODS
|
|
|
|
def make(original: Any, deep: bool) -> Any:
|
|
async def guarded(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
result = await original(*args, **kwargs)
|
|
except Exception as exc:
|
|
lic = _denial_license_error(denial_path)
|
|
if lic is not None:
|
|
raise lic from exc
|
|
raise
|
|
# See the sync variant: a denial can land while calls still
|
|
# succeed, so check the file after every call, not just on error.
|
|
lic = _denial_license_error(denial_path)
|
|
if lic is not None:
|
|
raise lic
|
|
if deep and result is not None:
|
|
_install_license_guard_async(result, denial_path)
|
|
return result
|
|
return guarded
|
|
|
|
# Wrap so the guard is not a descriptor (see sync variant, issue #488).
|
|
setattr(target, name, _Guarded(make(original, deep)))
|
|
|
|
|
|
# Sentinel to distinguish "viewport not provided" from "viewport=None" (disable emulation)
|
|
_VIEWPORT_UNSET = object()
|
|
|
|
|
|
def _default_no_viewport(browser: Any) -> None:
|
|
"""Default ``new_page()``/``new_context()`` to ``no_viewport=True``.
|
|
|
|
``launch()`` returns a raw Playwright ``Browser``; a bare ``browser.new_page()``
|
|
would otherwise inherit Playwright's emulated 1280x720 viewport, producing
|
|
``outerWidth < innerWidth`` — a physically impossible window (bot tell). We wrap
|
|
the two factory methods so pages track the real OS window instead. ``setdefault``
|
|
only: an explicit ``viewport`` or ``no_viewport`` from the caller is never
|
|
overridden (Playwright rejects passing both). Applied for headed launches only.
|
|
Composes under humanize's ``patch_browser`` (apply this first).
|
|
"""
|
|
orig_new_context = browser.new_context
|
|
orig_new_page = browser.new_page
|
|
|
|
def _patched_new_context(**kwargs: Any) -> Any:
|
|
if "viewport" not in kwargs:
|
|
kwargs.setdefault("no_viewport", True)
|
|
return orig_new_context(**kwargs)
|
|
|
|
def _patched_new_page(**kwargs: Any) -> Any:
|
|
if "viewport" not in kwargs:
|
|
kwargs.setdefault("no_viewport", True)
|
|
return orig_new_page(**kwargs)
|
|
|
|
browser.new_context = _patched_new_context
|
|
browser.new_page = _patched_new_page
|
|
|
|
|
|
def _default_no_viewport_async(browser: Any) -> None:
|
|
"""Async variant of :func:`_default_no_viewport`."""
|
|
orig_new_context = browser.new_context
|
|
orig_new_page = browser.new_page
|
|
|
|
async def _patched_new_context(**kwargs: Any) -> Any:
|
|
if "viewport" not in kwargs:
|
|
kwargs.setdefault("no_viewport", True)
|
|
return await orig_new_context(**kwargs)
|
|
|
|
async def _patched_new_page(**kwargs: Any) -> Any:
|
|
if "viewport" not in kwargs:
|
|
kwargs.setdefault("no_viewport", True)
|
|
return await orig_new_page(**kwargs)
|
|
|
|
browser.new_context = _patched_new_context
|
|
browser.new_page = _patched_new_page
|
|
|
|
|
|
def _resolve_context_viewport(
|
|
viewport: Any, headless: bool, headless_no_viewport: bool = False
|
|
) -> dict[str, Any]:
|
|
"""Return the viewport kwarg for a context.
|
|
|
|
Headed: no emulated viewport so the page tracks the real window. Headless on a
|
|
newer binary (``headless_no_viewport``): also ``no_viewport``, since it reports
|
|
coherent dimensions without emulation. Headless on an older binary: a fixed
|
|
``DEFAULT_VIEWPORT`` keeps dimensions coherent and deterministic. Explicit
|
|
``viewport`` / ``None`` honored.
|
|
"""
|
|
if viewport is _VIEWPORT_UNSET:
|
|
if headless and not headless_no_viewport:
|
|
return {"viewport": DEFAULT_VIEWPORT}
|
|
return {"no_viewport": True}
|
|
if viewport is None:
|
|
return {"no_viewport": True}
|
|
return {"viewport": viewport}
|
|
|
|
|
|
def _drop_conflicting_viewport(context_kwargs: dict[str, Any], kwargs: dict[str, Any]) -> None:
|
|
"""Playwright rejects passing both ``viewport`` and ``no_viewport``. ``viewport`` is a
|
|
named parameter (never in ``**kwargs``), so the only conflict is a caller passing
|
|
``no_viewport`` via ``**kwargs`` alongside an explicit ``viewport`` — the explicit
|
|
``no_viewport`` wins; drop the viewport so Playwright doesn't error.
|
|
"""
|
|
if "no_viewport" in kwargs and "viewport" in context_kwargs:
|
|
logger.debug("Both viewport and no_viewport requested; no_viewport (kwargs) wins")
|
|
context_kwargs.pop("viewport", None)
|
|
|
|
|
|
def _resolve_timezone(timezone: str | None, kwargs: dict[str, Any]) -> str | None:
|
|
"""Accept both timezone and timezone_id — either works, no warning."""
|
|
if "timezone_id" in kwargs:
|
|
if timezone is None:
|
|
timezone = kwargs.pop("timezone_id")
|
|
else:
|
|
kwargs.pop("timezone_id")
|
|
return timezone
|
|
|
|
|
|
def _check_removed_kwargs(kwargs: dict[str, Any]) -> None:
|
|
"""Raise a clear error for removed parameters that now fall into **kwargs."""
|
|
if "backend" in kwargs:
|
|
raise TypeError(
|
|
"The 'backend' parameter has been removed — patchright is no longer "
|
|
"supported and stock Playwright is the only backend. Remove the argument."
|
|
)
|
|
|
|
|
|
class _ProxySettingsRequired(TypedDict):
|
|
server: str
|
|
|
|
|
|
class ProxySettings(_ProxySettingsRequired, total=False):
|
|
"""Playwright-compatible proxy configuration."""
|
|
|
|
bypass: str
|
|
username: str
|
|
password: str
|
|
|
|
|
|
def launch(
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
timezone: str | None = None,
|
|
locale: str | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
_suppress_maximize: bool = False,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Launch stealth Chromium browser. Returns a Playwright Browser object.
|
|
|
|
Args:
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict.
|
|
String: 'http://user:pass@proxy:8080' (credentials auto-extracted).
|
|
Dict: {"server": "http://proxy:8080", "bypass": ".google.com", ...}
|
|
— passed directly to Playwright.
|
|
args: Additional Chromium CLI arguments to pass.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
Set to False if you want to pass your own --fingerprint flags.
|
|
timezone: IANA timezone (e.g. 'America/New_York'). Sets --fingerprint-timezone binary flag.
|
|
locale: BCP 47 locale (e.g. 'en-US'). Sets --lang binary flag.
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
Requires ``pip install cloakbrowser[geoip]``. Downloads ~70 MB
|
|
GeoLite2-City database on first use. Explicit timezone/locale
|
|
always override geoip results.
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed directly to playwright.chromium.launch().
|
|
|
|
Returns:
|
|
Playwright Browser object — use same API as playwright.chromium.launch().
|
|
|
|
Example:
|
|
>>> from cloakbrowser import launch
|
|
>>> browser = launch()
|
|
>>> page = browser.new_page()
|
|
>>> page.goto("https://bot.incolumitas.com")
|
|
>>> print(page.title())
|
|
>>> browser.close()
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
binary_path = ensure_binary(license_key=license_key, browser_version=browser_version, release_channel=release_channel)
|
|
timezone, locale, exit_ip = maybe_resolve_geoip(geoip, proxy, timezone, locale, args)
|
|
proxy_kwargs, proxy_extra_args = _resolve_proxy_config(proxy, browser_version, license_key, release_channel)
|
|
args = _resolve_webrtc_args(args, proxy)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
|
|
chrome_args = build_args(stealth_args, (args or []) + proxy_extra_args, timezone=timezone, locale=locale, headless=headless, extension_paths=extension_paths, start_maximized=binary_supports_maximized_window(license_key, browser_version, release_channel) and not _suppress_maximize)
|
|
_maybe_warn_windows_fonts(chrome_args)
|
|
|
|
logger.debug("Launching stealth Chromium (headless=%s, args=%d)", headless, len(chrome_args))
|
|
|
|
denial_path = mint_denial_file() if resolve_license_key(license_key) else None
|
|
launch_env = build_launch_env(license_key, user_env=kwargs.pop("env", None), status_file=denial_path)
|
|
env_kwargs = {} if launch_env is None else {"env": launch_env}
|
|
|
|
pw = sync_playwright().start()
|
|
try:
|
|
browser = pw.chromium.launch(
|
|
executable_path=binary_path,
|
|
headless=headless,
|
|
args=chrome_args,
|
|
ignore_default_args=IGNORE_DEFAULT_ARGS,
|
|
**env_kwargs,
|
|
**proxy_kwargs,
|
|
**kwargs,
|
|
)
|
|
except Exception as exc:
|
|
lic = _license_error(exc)
|
|
try:
|
|
pw.stop()
|
|
except Exception as stop_exc:
|
|
logger.warning("Playwright cleanup after launch failure failed: %s", stop_exc)
|
|
if lic is None:
|
|
raise
|
|
raise lic from exc
|
|
|
|
# Patch close() to also stop the Playwright instance
|
|
_original_close = browser.close
|
|
|
|
def _close_with_cleanup(*, reason: str | None = None) -> None:
|
|
try:
|
|
if reason is None:
|
|
_original_close()
|
|
else:
|
|
_original_close(reason=reason)
|
|
finally:
|
|
pw.stop()
|
|
|
|
browser.close = _close_with_cleanup
|
|
|
|
# Convert a post-handshake license denial into a clear error on the user's
|
|
# first call. Installed before the wraps below so it sits closest to the
|
|
# real call. Stash the path so launch_context() can guard its context too.
|
|
if denial_path:
|
|
browser._cloak_denial_path = denial_path
|
|
_install_license_guard(browser, denial_path)
|
|
|
|
# Proxy went to Chrome as --proxy-server, so Playwright's request client has no
|
|
# proxy (#579). The impl "context" event fires for every new context,
|
|
# including the one new_page() and launch_context() create internally.
|
|
if proxy_extra_args:
|
|
browser._impl_obj.on("context", _warn_on_request_client_use)
|
|
|
|
# Default new_page()/new_context() to no_viewport for headed (page tracks the
|
|
# real window) and for headless on binaries that report coherent dimensions
|
|
# natively; older headless binaries keep Playwright's default viewport. Apply
|
|
# before humanize so the wraps compose.
|
|
if not headless or binary_supports_headless_no_viewport(license_key, browser_version, release_channel):
|
|
_default_no_viewport(browser)
|
|
|
|
# Human-like behavioral patching
|
|
if humanize:
|
|
from .human import patch_browser
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_browser(browser, cfg)
|
|
|
|
return browser
|
|
|
|
|
|
async def launch_async( # noqa: C901
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
timezone: str | None = None,
|
|
locale: str | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
_suppress_maximize: bool = False,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Async version of launch(). Returns a Playwright Browser object.
|
|
|
|
Args:
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict (see launch() for details).
|
|
args: Additional Chromium CLI arguments to pass.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
timezone: IANA timezone (e.g. 'America/New_York'). Sets --fingerprint-timezone binary flag.
|
|
locale: BCP 47 locale (e.g. 'en-US'). Sets --lang binary flag.
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed directly to playwright.chromium.launch().
|
|
|
|
Returns:
|
|
Playwright Browser object (async API).
|
|
|
|
Example:
|
|
>>> import asyncio
|
|
>>> from cloakbrowser import launch_async
|
|
>>>
|
|
>>> async def main():
|
|
... browser = await launch_async()
|
|
... page = await browser.new_page()
|
|
... await page.goto("https://bot.incolumitas.com")
|
|
... print(await page.title())
|
|
... await browser.close()
|
|
>>>
|
|
>>> asyncio.run(main())
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
from playwright.async_api import async_playwright
|
|
|
|
# Blocking network calls (license/download, exit-IP lookups) run in a thread so
|
|
# they don't freeze the caller's event loop.
|
|
binary_path = await asyncio.to_thread(ensure_binary, license_key=license_key, browser_version=browser_version, release_channel=release_channel)
|
|
timezone, locale, exit_ip = await asyncio.to_thread(maybe_resolve_geoip, geoip, proxy, timezone, locale, args)
|
|
proxy_kwargs, proxy_extra_args = _resolve_proxy_config(proxy, browser_version, license_key, release_channel)
|
|
args = await asyncio.to_thread(_resolve_webrtc_args, args, proxy)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
chrome_args = build_args(stealth_args, (args or []) + proxy_extra_args, timezone=timezone, locale=locale, headless=headless, extension_paths=extension_paths, start_maximized=binary_supports_maximized_window(license_key, browser_version, release_channel) and not _suppress_maximize)
|
|
_maybe_warn_windows_fonts(chrome_args)
|
|
|
|
logger.debug("Launching stealth Chromium async (headless=%s, args=%d)", headless, len(chrome_args))
|
|
|
|
denial_path = mint_denial_file() if resolve_license_key(license_key) else None
|
|
launch_env = build_launch_env(license_key, user_env=kwargs.pop("env", None), status_file=denial_path)
|
|
env_kwargs = {} if launch_env is None else {"env": launch_env}
|
|
|
|
pw = await async_playwright().start()
|
|
try:
|
|
browser = await pw.chromium.launch(
|
|
executable_path=binary_path,
|
|
headless=headless,
|
|
args=chrome_args,
|
|
ignore_default_args=IGNORE_DEFAULT_ARGS,
|
|
**env_kwargs,
|
|
**proxy_kwargs,
|
|
**kwargs,
|
|
)
|
|
except Exception as exc:
|
|
lic = _license_error(exc)
|
|
try:
|
|
await pw.stop()
|
|
except Exception as stop_exc:
|
|
logger.warning("Playwright cleanup after launch failure failed: %s", stop_exc)
|
|
if lic is None:
|
|
raise
|
|
raise lic from exc
|
|
|
|
# Patch close() to also stop the Playwright instance
|
|
_original_close = browser.close
|
|
|
|
async def _close_with_cleanup(*, reason: str | None = None) -> None:
|
|
try:
|
|
if reason is None:
|
|
await _original_close()
|
|
else:
|
|
await _original_close(reason=reason)
|
|
finally:
|
|
await pw.stop()
|
|
|
|
browser.close = _close_with_cleanup
|
|
|
|
# Convert a post-handshake license denial into a clear error on first use
|
|
# (see launch()). Stash the path for launch_context_async().
|
|
if denial_path:
|
|
browser._cloak_denial_path = denial_path
|
|
_install_license_guard_async(browser, denial_path)
|
|
|
|
# Warn on first use of Playwright's request client (see launch(), #579).
|
|
if proxy_extra_args:
|
|
browser._impl_obj.on("context", _warn_on_request_client_use)
|
|
|
|
# Default new_page()/new_context() to no_viewport for headed and qualifying
|
|
# headless binaries (see launch()).
|
|
if not headless or binary_supports_headless_no_viewport(license_key, browser_version, release_channel):
|
|
_default_no_viewport_async(browser)
|
|
|
|
# Human-like behavioral patching (async variant)
|
|
if humanize:
|
|
from .human import patch_browser_async
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_browser_async(browser, cfg)
|
|
|
|
return browser
|
|
|
|
|
|
def launch_persistent_context(
|
|
user_data_dir: str | os.PathLike,
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
user_agent: str | None = None,
|
|
viewport: Any = _VIEWPORT_UNSET,
|
|
locale: str | None = None,
|
|
timezone: str | None = None,
|
|
color_scheme: Literal["light", "dark", "no-preference"] | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Launch stealth browser with a persistent profile and return a BrowserContext.
|
|
|
|
This persists cookies, localStorage, cache, and other browser state across
|
|
sessions by storing them in ``user_data_dir``. Also avoids incognito detection
|
|
by services like BrowserScan (-10% penalty).
|
|
|
|
Args:
|
|
user_data_dir: Path to the directory where browser profile data is stored.
|
|
Created automatically if it doesn't exist. Reuse the same path across
|
|
sessions to restore cookies, localStorage, cached credentials, etc.
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict (see launch() for details).
|
|
args: Additional Chromium CLI arguments.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
user_agent: Custom user agent string.
|
|
viewport: Viewport size dict, e.g. {"width": 1920, "height": 1080}.
|
|
Pass None to disable viewport emulation (use OS window size).
|
|
locale: Browser locale, e.g. "en-US".
|
|
timezone: IANA timezone (e.g. 'America/New_York').
|
|
color_scheme: Color scheme preference — 'light', 'dark', or 'no-preference'.
|
|
Default: None (uses Chromium default, which is 'light').
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
Requires ``pip install cloakbrowser[geoip]``.
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed directly to playwright.chromium.launch_persistent_context().
|
|
|
|
Returns:
|
|
Playwright BrowserContext object backed by a persistent profile.
|
|
Call ``.close()`` when done — this also stops the Playwright instance.
|
|
|
|
Example:
|
|
>>> from cloakbrowser import launch_persistent_context
|
|
>>> ctx = launch_persistent_context("./my-profile", headless=False)
|
|
>>> page = ctx.new_page()
|
|
>>> page.goto("https://protected-site.com")
|
|
>>> ctx.close() # Profile is saved; re-use path next run to restore state.
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
timezone = _resolve_timezone(timezone, kwargs)
|
|
|
|
binary_path = ensure_binary(license_key=license_key, browser_version=browser_version, release_channel=release_channel)
|
|
timezone, locale, exit_ip = maybe_resolve_geoip(geoip, proxy, timezone, locale, args)
|
|
proxy_kwargs, proxy_extra_args = _resolve_proxy_config(proxy, browser_version, license_key, release_channel)
|
|
args = _resolve_webrtc_args(args, proxy)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
args = persistent_seed_args(user_data_dir, stealth_args, args)
|
|
chrome_args = build_args(stealth_args, (args or []) + proxy_extra_args, timezone=timezone, locale=locale, headless=headless, extension_paths=extension_paths, start_maximized=binary_supports_maximized_window(license_key, browser_version, release_channel) and viewport is _VIEWPORT_UNSET and "viewport" not in kwargs and "no_viewport" not in kwargs)
|
|
_maybe_warn_windows_fonts(chrome_args)
|
|
|
|
logger.debug(
|
|
"Launching persistent stealth Chromium (headless=%s, user_data_dir=%s)",
|
|
headless,
|
|
user_data_dir,
|
|
)
|
|
|
|
# locale and timezone are set via binary flags (--lang, --fingerprint-timezone)
|
|
# — NOT via Playwright context kwargs which use detectable CDP emulation.
|
|
context_kwargs: dict[str, Any] = {}
|
|
if user_agent:
|
|
context_kwargs["user_agent"] = user_agent
|
|
context_kwargs.update(
|
|
_resolve_context_viewport(
|
|
viewport, headless, binary_supports_headless_no_viewport(license_key, browser_version, release_channel)
|
|
)
|
|
)
|
|
if color_scheme:
|
|
context_kwargs["color_scheme"] = color_scheme
|
|
context_kwargs.update(kwargs)
|
|
_drop_conflicting_viewport(context_kwargs, kwargs)
|
|
|
|
# Resolve env for the browser process (license key injection, if needed)
|
|
user_env = context_kwargs.pop("env", None)
|
|
denial_path = mint_denial_file() if resolve_license_key(license_key) else None
|
|
launch_env = build_launch_env(license_key, user_env=user_env, status_file=denial_path)
|
|
if launch_env is not None:
|
|
context_kwargs["env"] = launch_env
|
|
|
|
seed_widevine_hint(user_data_dir, binary_path)
|
|
|
|
pw = sync_playwright().start()
|
|
try:
|
|
context = pw.chromium.launch_persistent_context(
|
|
user_data_dir=os.fspath(user_data_dir),
|
|
executable_path=binary_path,
|
|
headless=headless,
|
|
args=chrome_args,
|
|
ignore_default_args=IGNORE_DEFAULT_ARGS,
|
|
**proxy_kwargs,
|
|
**context_kwargs,
|
|
)
|
|
except Exception as exc:
|
|
lic = _license_error(exc)
|
|
try:
|
|
pw.stop()
|
|
except Exception as stop_exc:
|
|
logger.warning("Playwright cleanup after launch failure failed: %s", stop_exc)
|
|
if lic is None:
|
|
raise
|
|
raise lic from exc
|
|
|
|
# Patch close() to also stop the Playwright instance
|
|
_original_close = context.close
|
|
|
|
def _close_with_cleanup(*, reason: str | None = None) -> None:
|
|
try:
|
|
if reason is None:
|
|
_original_close()
|
|
else:
|
|
_original_close(reason=reason)
|
|
finally:
|
|
pw.stop()
|
|
|
|
context.close = _close_with_cleanup
|
|
|
|
# Warn on first use of Playwright's request client (see launch(), #579).
|
|
if proxy_extra_args:
|
|
_warn_on_request_client_use(context._impl_obj)
|
|
|
|
# Human-like behavioral patching. Applied BEFORE the license guard: the guard
|
|
# stores its wrapper on the page/context object itself (it needs this launch's
|
|
# denial path), and an object attribute shadows the class method humanize
|
|
# patches. Guarding first would therefore capture Playwright's own methods and
|
|
# silently disable humanize for every call on that page.
|
|
if humanize:
|
|
from .human import patch_context
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_context(context, cfg)
|
|
|
|
# The persistent path hands back a context, so guard it (new_page deep — see
|
|
# launch()). A persistent context also arrives with a page already open, so
|
|
# the user navigates pages[0] directly and never calls new_page — guard those
|
|
# existing pages too, or a post-handshake denial surfaces as a bare
|
|
# TargetClosedError on that first navigation.
|
|
if denial_path:
|
|
_install_license_guard(context, denial_path)
|
|
for _pg in context.pages:
|
|
_install_license_guard(_pg, denial_path)
|
|
|
|
return context
|
|
|
|
|
|
async def launch_persistent_context_async(
|
|
user_data_dir: str | os.PathLike,
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
user_agent: str | None = None,
|
|
viewport: Any = _VIEWPORT_UNSET,
|
|
locale: str | None = None,
|
|
timezone: str | None = None,
|
|
color_scheme: Literal["light", "dark", "no-preference"] | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Async version of launch_persistent_context().
|
|
|
|
Launch stealth browser with a persistent profile and return a BrowserContext.
|
|
This persists cookies, localStorage, cache, and other browser state across
|
|
sessions by storing them in ``user_data_dir``.
|
|
|
|
Args:
|
|
user_data_dir: Path to the directory where browser profile data is stored.
|
|
Created automatically if it doesn't exist.
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict (see launch() for details).
|
|
args: Additional Chromium CLI arguments.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
user_agent: Custom user agent string.
|
|
viewport: Viewport size dict, e.g. {"width": 1920, "height": 1080}.
|
|
Pass None to disable viewport emulation (use OS window size).
|
|
locale: Browser locale, e.g. "en-US".
|
|
timezone: IANA timezone (e.g. 'America/New_York').
|
|
color_scheme: Color scheme preference — 'light', 'dark', or 'no-preference'.
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed directly to playwright.chromium.launch_persistent_context().
|
|
|
|
Returns:
|
|
Playwright BrowserContext object backed by a persistent profile (async API).
|
|
Call ``await .close()`` when done.
|
|
|
|
Example:
|
|
>>> import asyncio
|
|
>>> from cloakbrowser import launch_persistent_context_async
|
|
>>>
|
|
>>> async def main():
|
|
... ctx = await launch_persistent_context_async("./my-profile", headless=False)
|
|
... page = await ctx.new_page()
|
|
... await page.goto("https://protected-site.com")
|
|
... await ctx.close()
|
|
>>>
|
|
>>> asyncio.run(main())
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
from playwright.async_api import async_playwright
|
|
|
|
timezone = _resolve_timezone(timezone, kwargs)
|
|
|
|
# Blocking network calls run in a thread (see launch_async).
|
|
binary_path = await asyncio.to_thread(ensure_binary, license_key=license_key, browser_version=browser_version, release_channel=release_channel)
|
|
timezone, locale, exit_ip = await asyncio.to_thread(maybe_resolve_geoip, geoip, proxy, timezone, locale, args)
|
|
proxy_kwargs, proxy_extra_args = _resolve_proxy_config(proxy, browser_version, license_key, release_channel)
|
|
args = await asyncio.to_thread(_resolve_webrtc_args, args, proxy)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
args = persistent_seed_args(user_data_dir, stealth_args, args)
|
|
chrome_args = build_args(stealth_args, (args or []) + proxy_extra_args, timezone=timezone, locale=locale, headless=headless, extension_paths=extension_paths, start_maximized=binary_supports_maximized_window(license_key, browser_version, release_channel) and viewport is _VIEWPORT_UNSET and "viewport" not in kwargs and "no_viewport" not in kwargs)
|
|
_maybe_warn_windows_fonts(chrome_args)
|
|
|
|
logger.debug(
|
|
"Launching persistent stealth Chromium async (headless=%s, user_data_dir=%s)",
|
|
headless,
|
|
user_data_dir,
|
|
)
|
|
|
|
# locale and timezone are set via binary flags (--lang, --fingerprint-timezone)
|
|
# — NOT via Playwright context kwargs which use detectable CDP emulation.
|
|
context_kwargs: dict[str, Any] = {}
|
|
if user_agent:
|
|
context_kwargs["user_agent"] = user_agent
|
|
context_kwargs.update(
|
|
_resolve_context_viewport(
|
|
viewport, headless, binary_supports_headless_no_viewport(license_key, browser_version, release_channel)
|
|
)
|
|
)
|
|
if color_scheme:
|
|
context_kwargs["color_scheme"] = color_scheme
|
|
context_kwargs.update(kwargs)
|
|
_drop_conflicting_viewport(context_kwargs, kwargs)
|
|
|
|
# Resolve env for the browser process (license key injection, if needed)
|
|
user_env = context_kwargs.pop("env", None)
|
|
denial_path = mint_denial_file() if resolve_license_key(license_key) else None
|
|
launch_env = build_launch_env(license_key, user_env=user_env, status_file=denial_path)
|
|
if launch_env is not None:
|
|
context_kwargs["env"] = launch_env
|
|
|
|
seed_widevine_hint(user_data_dir, binary_path)
|
|
|
|
pw = await async_playwright().start()
|
|
try:
|
|
context = await pw.chromium.launch_persistent_context(
|
|
user_data_dir=os.fspath(user_data_dir),
|
|
executable_path=binary_path,
|
|
headless=headless,
|
|
args=chrome_args,
|
|
ignore_default_args=IGNORE_DEFAULT_ARGS,
|
|
**proxy_kwargs,
|
|
**context_kwargs,
|
|
)
|
|
except Exception as exc:
|
|
lic = _license_error(exc)
|
|
try:
|
|
await pw.stop()
|
|
except Exception as stop_exc:
|
|
logger.warning("Playwright cleanup after launch failure failed: %s", stop_exc)
|
|
if lic is None:
|
|
raise
|
|
raise lic from exc
|
|
|
|
# Patch close() to also stop the Playwright instance
|
|
_original_close = context.close
|
|
|
|
async def _close_with_cleanup(*, reason: str | None = None) -> None:
|
|
try:
|
|
if reason is None:
|
|
await _original_close()
|
|
else:
|
|
await _original_close(reason=reason)
|
|
finally:
|
|
await pw.stop()
|
|
|
|
context.close = _close_with_cleanup
|
|
|
|
# Warn on first use of Playwright's request client (see launch(), #579).
|
|
if proxy_extra_args:
|
|
_warn_on_request_client_use(context._impl_obj)
|
|
|
|
# Human-like behavioral patching (async variant). Applied BEFORE the license
|
|
# guard, for the same reason as the sync variant: the guard stores its wrapper
|
|
# on the page/context object, which shadows the class method humanize patches.
|
|
if humanize:
|
|
from .human import patch_context_async
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_context_async(context, cfg)
|
|
|
|
# The persistent path hands back a context, so guard it (new_page deep — see
|
|
# launch()). A persistent context also arrives with a page already open (see
|
|
# the sync variant): guard those existing pages so a post-handshake denial on
|
|
# pages[0] surfaces cleanly instead of as a bare TargetClosedError.
|
|
if denial_path:
|
|
context._cloak_denial_path = denial_path
|
|
_install_license_guard_async(context, denial_path)
|
|
for _pg in context.pages:
|
|
_install_license_guard_async(_pg, denial_path)
|
|
|
|
return context
|
|
|
|
|
|
def launch_context(
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
user_agent: str | None = None,
|
|
viewport: Any = _VIEWPORT_UNSET,
|
|
locale: str | None = None,
|
|
timezone: str | None = None,
|
|
color_scheme: Literal["light", "dark", "no-preference"] | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Launch stealth browser and return a BrowserContext with common options pre-set.
|
|
|
|
Convenience function that creates a browser + context in one call.
|
|
Useful for setting user agent, viewport, locale, etc.
|
|
|
|
Args:
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict (see launch() for details).
|
|
args: Additional Chromium CLI arguments.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
user_agent: Custom user agent string.
|
|
viewport: Viewport size dict, e.g. {"width": 1920, "height": 1080}.
|
|
Pass None to disable viewport emulation (use OS window size).
|
|
locale: Browser locale, e.g. "en-US".
|
|
timezone: IANA timezone (e.g. 'America/New_York').
|
|
color_scheme: Color scheme preference — 'light', 'dark', or 'no-preference'.
|
|
Default: None (uses Chromium default, which is 'light').
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed to browser.new_context().
|
|
|
|
Returns:
|
|
Playwright BrowserContext object.
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
timezone = _resolve_timezone(timezone, kwargs)
|
|
|
|
# Resolve geoip BEFORE launch() to avoid double-resolution and ensure
|
|
# resolved values flow to binary flags
|
|
timezone, locale, exit_ip = maybe_resolve_geoip(geoip, proxy, timezone, locale, args)
|
|
# Inject geoip exit IP for WebRTC spoofing (free — no extra HTTP call)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
# --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.
|
|
browser = launch(headless=headless, proxy=proxy, args=args, stealth_args=stealth_args,
|
|
timezone=timezone, locale=locale, extension_paths=extension_paths,
|
|
license_key=license_key, browser_version=browser_version,
|
|
release_channel=release_channel,
|
|
# Caller chose a viewport geometry → don't also auto-maximize
|
|
# the window (mirrors the persistent-context path + JS).
|
|
_suppress_maximize=(viewport is not _VIEWPORT_UNSET or "no_viewport" in kwargs))
|
|
|
|
context_kwargs: dict[str, Any] = {}
|
|
if user_agent:
|
|
context_kwargs["user_agent"] = user_agent
|
|
context_kwargs.update(
|
|
_resolve_context_viewport(
|
|
viewport, headless, binary_supports_headless_no_viewport(license_key, browser_version, release_channel)
|
|
)
|
|
)
|
|
if color_scheme:
|
|
context_kwargs["color_scheme"] = color_scheme
|
|
context_kwargs.update(kwargs)
|
|
_drop_conflicting_viewport(context_kwargs, kwargs)
|
|
|
|
try:
|
|
context = browser.new_context(**context_kwargs)
|
|
except Exception:
|
|
browser.close()
|
|
raise
|
|
|
|
# Patch close() to also close the browser (and its Playwright instance)
|
|
_original_ctx_close = context.close
|
|
|
|
def _close_context_with_cleanup() -> None:
|
|
try:
|
|
_original_ctx_close()
|
|
finally:
|
|
browser.close()
|
|
|
|
context.close = _close_context_with_cleanup
|
|
|
|
# browser.new_context above is already guarded by launch(); also guard the
|
|
# context's new_page for a denial that lands after the context is created.
|
|
denial_path = getattr(browser, "_cloak_denial_path", None)
|
|
if denial_path:
|
|
_install_license_guard(context, denial_path)
|
|
|
|
# Human-like behavioral patching
|
|
if humanize:
|
|
from .human import patch_context
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_context(context, cfg)
|
|
|
|
return context
|
|
|
|
|
|
async def launch_context_async(
|
|
headless: bool = True,
|
|
proxy: str | ProxySettings | None = None,
|
|
args: list[str] | None = None,
|
|
stealth_args: bool = True,
|
|
user_agent: str | None = None,
|
|
viewport: Any = _VIEWPORT_UNSET,
|
|
locale: str | None = None,
|
|
timezone: str | None = None,
|
|
color_scheme: Literal["light", "dark", "no-preference"] | None = None,
|
|
geoip: bool = False,
|
|
humanize: bool = False,
|
|
human_preset: HumanPreset = "default",
|
|
human_config: HumanConfigOverrides | None = None,
|
|
extension_paths: list[str] | None = None,
|
|
license_key: str | None = None,
|
|
browser_version: str | None = None,
|
|
release_channel: str | None = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
"""Async version of launch_context().
|
|
|
|
Launch stealth browser and return a BrowserContext with common options pre-set.
|
|
All extra kwargs are forwarded to ``browser.new_context()`` — use this for
|
|
``storage_state``, ``permissions``, ``extra_http_headers``, etc. without needing
|
|
a persistent profile folder.
|
|
|
|
Args:
|
|
headless: Run in headless mode (default True).
|
|
proxy: Proxy URL string or Playwright proxy dict (see launch() for details).
|
|
args: Additional Chromium CLI arguments.
|
|
extension_paths: List of Chrome extension paths to load.
|
|
stealth_args: Include default stealth fingerprint args (default True).
|
|
user_agent: Custom user agent string.
|
|
viewport: Viewport size dict, e.g. {"width": 1920, "height": 1080}.
|
|
Pass None to disable viewport emulation (use OS window size).
|
|
locale: Browser locale, e.g. "en-US".
|
|
timezone: IANA timezone (e.g. 'America/New_York').
|
|
color_scheme: Color scheme preference — 'light', 'dark', or 'no-preference'.
|
|
geoip: Auto-detect timezone/locale from proxy IP (default False).
|
|
humanize: Enable human-like mouse, keyboard, scroll behavior (default False).
|
|
human_preset: Humanize preset — 'default' or 'careful' (default 'default').
|
|
human_config: Custom humanize config mapping to override preset values.
|
|
**kwargs: Passed to browser.new_context() — e.g. storage_state, permissions.
|
|
|
|
Returns:
|
|
Playwright BrowserContext object (async API).
|
|
Call ``await .close()`` when done — this also closes the underlying browser.
|
|
|
|
Example:
|
|
>>> import asyncio
|
|
>>> from cloakbrowser import launch_context_async
|
|
>>>
|
|
>>> async def main():
|
|
... # Load saved session (cookies, localStorage)
|
|
... ctx = await launch_context_async(
|
|
... headless=True,
|
|
... storage_state="state.json",
|
|
... )
|
|
... page = await ctx.new_page()
|
|
... await page.goto("https://example.com")
|
|
... # Save state back
|
|
... await ctx.storage_state(path="state.json")
|
|
... await ctx.close()
|
|
>>>
|
|
>>> asyncio.run(main())
|
|
"""
|
|
_check_removed_kwargs(kwargs)
|
|
|
|
timezone = _resolve_timezone(timezone, kwargs)
|
|
|
|
# Resolve geoip BEFORE launch_async() to avoid double-resolution and ensure
|
|
# resolved values flow to binary flags. Threaded: the lookup is blocking network I/O.
|
|
timezone, locale, exit_ip = await asyncio.to_thread(maybe_resolve_geoip, geoip, proxy, timezone, locale, args)
|
|
args = _append_webrtc_exit_ip(args, exit_ip)
|
|
# --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.
|
|
browser = await launch_async(headless=headless, proxy=proxy, args=args, stealth_args=stealth_args,
|
|
timezone=timezone, locale=locale, extension_paths=extension_paths,
|
|
license_key=license_key, browser_version=browser_version,
|
|
release_channel=release_channel,
|
|
# Caller chose a viewport geometry → don't also auto-maximize
|
|
# the window (mirrors the persistent-context path + JS).
|
|
_suppress_maximize=(viewport is not _VIEWPORT_UNSET or "no_viewport" in kwargs))
|
|
|
|
context_kwargs: dict[str, Any] = {}
|
|
if user_agent:
|
|
context_kwargs["user_agent"] = user_agent
|
|
context_kwargs.update(
|
|
_resolve_context_viewport(
|
|
viewport, headless, binary_supports_headless_no_viewport(license_key, browser_version, release_channel)
|
|
)
|
|
)
|
|
if color_scheme:
|
|
context_kwargs["color_scheme"] = color_scheme
|
|
context_kwargs.update(kwargs)
|
|
_drop_conflicting_viewport(context_kwargs, kwargs)
|
|
|
|
# Catch BaseException (not just Exception) so that asyncio.CancelledError
|
|
# triggers browser cleanup — otherwise the underlying Chromium process
|
|
# leaks when the awaiting task is cancelled.
|
|
try:
|
|
context = await browser.new_context(**context_kwargs)
|
|
except BaseException:
|
|
try:
|
|
await browser.close()
|
|
except BaseException:
|
|
pass
|
|
raise
|
|
|
|
# Patch close() to also close the browser (and its Playwright instance)
|
|
_original_ctx_close = context.close
|
|
|
|
async def _close_context_with_cleanup() -> None:
|
|
try:
|
|
await _original_ctx_close()
|
|
finally:
|
|
await browser.close()
|
|
|
|
context.close = _close_context_with_cleanup
|
|
|
|
# browser.new_context above is already guarded by launch_async(); also guard
|
|
# the context's new_page for a denial that lands after context creation.
|
|
denial_path = getattr(browser, "_cloak_denial_path", None)
|
|
if denial_path:
|
|
_install_license_guard_async(context, denial_path)
|
|
|
|
# Human-like behavioral patching (async variant)
|
|
if humanize:
|
|
from .human import patch_context_async
|
|
from .human.config import resolve_config
|
|
cfg = resolve_config(human_preset, human_config)
|
|
patch_context_async(context, cfg)
|
|
|
|
return context
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ensure_proxy_scheme(proxy_url: str) -> str:
|
|
"""Prepend http:// to schemeless proxy URLs so parsers can extract hostname."""
|
|
return proxy_url if "://" in proxy_url else f"http://{proxy_url}"
|
|
|
|
|
|
def _assemble_proxy_url(
|
|
scheme: str,
|
|
host: str,
|
|
port: int | None,
|
|
enc_user: str,
|
|
enc_pass: str | None,
|
|
path: str = "",
|
|
params: str = "",
|
|
query: str = "",
|
|
fragment: str = "",
|
|
) -> str:
|
|
"""Build a proxy URL from already-percent-encoded credentials and host parts.
|
|
|
|
``enc_pass is None`` means no password (no colon in userinfo). Empty string
|
|
means present-but-empty (colon preserved). This mirrors the distinction
|
|
urlparse makes between ``user@host`` and ``user:@host``.
|
|
"""
|
|
if ":" in host: # IPv6 literal — re-add brackets
|
|
host = f"[{host}]"
|
|
if enc_pass is not None:
|
|
userinfo = f"{enc_user}:{enc_pass}@"
|
|
elif enc_user:
|
|
userinfo = f"{enc_user}@"
|
|
else:
|
|
userinfo = ""
|
|
netloc = f"{userinfo}{host}"
|
|
if port is not None:
|
|
netloc += f":{port}"
|
|
return urlunparse((scheme, netloc, path, params, query, fragment))
|
|
|
|
|
|
def _reconstruct_socks_url(proxy: ProxySettings) -> str:
|
|
"""Reconstruct a SOCKS5 URL with inline credentials from a Playwright proxy dict."""
|
|
server = proxy.get("server", "")
|
|
username = proxy.get("username", "")
|
|
password = proxy.get("password", "")
|
|
if not username:
|
|
return server
|
|
parsed = urlparse(server)
|
|
enc_user = quote(username, safe="")
|
|
# Dict convention: empty/missing password → no colon.
|
|
enc_pass = quote(password, safe="") if password else None
|
|
return _assemble_proxy_url(
|
|
parsed.scheme, parsed.hostname or "", parsed.port,
|
|
enc_user, enc_pass, parsed.path,
|
|
)
|
|
|
|
|
|
def _normalize_socks_string_url(url: str) -> str:
|
|
"""Re-encode credentials in a SOCKS5 URL string so Chromium's parser doesn't
|
|
truncate them at special chars like '='. Idempotent: pre-encoded input stays
|
|
the same (decoded then re-encoded).
|
|
|
|
Emits an INFO log when re-encoding actually changes the URL, so users who
|
|
previously hit silent SOCKS5 fallback (#157) can see what the wrapper did.
|
|
Silent on already-encoded inputs (no false-positive noise).
|
|
|
|
On unparseable input (invalid port, broken IPv6 literal, etc.) logs a
|
|
warning and returns the original string — preserves pre-fix pass-through
|
|
behavior so Chromium's own error handling kicks in.
|
|
"""
|
|
try:
|
|
parsed = urlparse(url)
|
|
# Accessing .port raises ValueError on invalid port strings.
|
|
_ = parsed.port
|
|
except ValueError as e:
|
|
logger.warning("Malformed SOCKS5 proxy URL, passing through unchanged: %s", e)
|
|
return url
|
|
# Skip only if no credentials at all (username AND password both absent).
|
|
# urlparse returns None for absent components, "" for present-but-empty.
|
|
if parsed.username is None and parsed.password is None:
|
|
return url
|
|
raw_user = parsed.username or ""
|
|
enc_user = quote(unquote(raw_user), safe="") if raw_user else ""
|
|
# Preserve the colon separator when password component is present, even if
|
|
# empty, so `user:@host` stays `user:@host`.
|
|
if parsed.password is not None:
|
|
raw_pass = parsed.password
|
|
enc_pass = quote(unquote(raw_pass), safe="") if raw_pass else ""
|
|
else:
|
|
raw_pass = None
|
|
enc_pass = None
|
|
normalized = _assemble_proxy_url(
|
|
parsed.scheme, parsed.hostname or "", parsed.port,
|
|
enc_user, enc_pass,
|
|
parsed.path, parsed.params, parsed.query, parsed.fragment,
|
|
)
|
|
# Compare credentials, not the full URL: urlparse cosmetically lowercases
|
|
# scheme and hostname, so a full-string compare would falsely fire on
|
|
# `socks5://USER:pass@HOST.com:1080` even when no encoding work happened.
|
|
if enc_user != raw_user or enc_pass != raw_pass:
|
|
logger.info(
|
|
"Auto URL-encoded SOCKS5 proxy credentials (special characters "
|
|
"detected). Pre-encode the URL to suppress this notice."
|
|
)
|
|
return normalized
|
|
|
|
|
|
def _extract_proxy_url(proxy: str | ProxySettings | None) -> str | None:
|
|
"""Extract and normalize proxy URL string from proxy param.
|
|
|
|
For proxy dicts with separate username/password fields, reconstructs
|
|
the full URL with inline credentials so proxy auth works.
|
|
"""
|
|
if proxy is None:
|
|
return None
|
|
if isinstance(proxy, dict):
|
|
server = proxy.get("server", "")
|
|
if not server:
|
|
return None
|
|
if proxy.get("username"):
|
|
return (
|
|
_reconstruct_socks_url(proxy) if _is_socks_proxy(proxy)
|
|
else _reconstruct_http_url(proxy)
|
|
)
|
|
return _ensure_proxy_scheme(server)
|
|
return _ensure_proxy_scheme(proxy)
|
|
|
|
|
|
def _get_flag_value(args: list[str] | None, *keys: str) -> str | None:
|
|
"""Return the value of the first ``--key=value`` flag found in *args*, else None."""
|
|
for a in args or []:
|
|
for k in keys:
|
|
if a.startswith(k + "="):
|
|
return a.split("=", 1)[1]
|
|
return None
|
|
|
|
|
|
def maybe_resolve_geoip(
|
|
geoip: bool,
|
|
proxy: str | ProxySettings | None,
|
|
timezone: str | None,
|
|
locale: str | None,
|
|
args: list[str] | None = None,
|
|
) -> tuple[str | None, str | None, str | None]:
|
|
"""Auto-fill timezone/locale from the egress IP when geoip is enabled.
|
|
|
|
Returns ``(timezone, locale, exit_ip)``. *exit_ip* is a free bonus
|
|
from the geoip lookup (no extra HTTP call) — used for WebRTC spoofing.
|
|
|
|
With a proxy the egress IP is the proxy's exit IP; with no proxy it is
|
|
the machine's own public IP, so geoip works proxy-free too.
|
|
|
|
A timezone/locale set as a raw flag in *args* (``--fingerprint-timezone``,
|
|
``--lang``, ``--fingerprint-locale``) counts as explicit: it is promoted to
|
|
the param so geoip leaves it alone, mirroring the ``timezone=``/``locale=``
|
|
override path.
|
|
"""
|
|
if not geoip:
|
|
return timezone, locale, None
|
|
|
|
# Promote raw flags to explicit params so geoip doesn't clobber them.
|
|
if timezone is None:
|
|
timezone = _get_flag_value(args, "--fingerprint-timezone")
|
|
if locale is None:
|
|
locale = _get_flag_value(args, "--lang", "--fingerprint-locale")
|
|
|
|
from .geoip import resolve_proxy_exit_ip, resolve_proxy_geo_with_ip
|
|
|
|
# None when no proxy → echo services resolve the machine's own public IP
|
|
proxy_url = _extract_proxy_url(proxy) if proxy else None
|
|
|
|
# When both tz/locale are explicit, resolve the exit IP for WebRTC — but only
|
|
# with a proxy. With no proxy the WebRTC IP would just be the real connection
|
|
# IP the site already sees (a no-op), so skip the third-party echo call.
|
|
if timezone is not None and locale is not None:
|
|
exit_ip = resolve_proxy_exit_ip(proxy_url) if proxy_url else None
|
|
return timezone, locale, exit_ip
|
|
|
|
geo_tz, geo_locale, exit_ip = resolve_proxy_geo_with_ip(proxy_url)
|
|
if timezone is None:
|
|
timezone = geo_tz
|
|
if locale is None:
|
|
locale = geo_locale
|
|
missing = [name for name, value in (("timezone", timezone), ("locale", locale)) if value is None]
|
|
if missing:
|
|
raise RuntimeError(
|
|
"GeoIP resolution failed: could not determine " + " and ".join(missing)
|
|
)
|
|
return timezone, locale, exit_ip
|
|
|
|
|
|
def _resolve_webrtc_args(
|
|
args: list[str] | None,
|
|
proxy: str | ProxySettings | None,
|
|
) -> list[str] | None:
|
|
"""Replace --fingerprint-webrtc-ip=auto with the resolved proxy exit IP.
|
|
|
|
Returns args unchanged if no ``auto`` value is present.
|
|
"""
|
|
if not args:
|
|
return args
|
|
idx = None
|
|
for i, a in enumerate(args):
|
|
if a == "--fingerprint-webrtc-ip=auto":
|
|
idx = i
|
|
break
|
|
if idx is None:
|
|
return args
|
|
proxy_url = _extract_proxy_url(proxy)
|
|
if not proxy_url:
|
|
logger.warning("--fingerprint-webrtc-ip=auto requires a proxy; removing flag")
|
|
args = list(args)
|
|
del args[idx]
|
|
return args
|
|
try:
|
|
from .geoip import resolve_proxy_exit_ip
|
|
exit_ip = resolve_proxy_exit_ip(proxy_url)
|
|
except Exception:
|
|
logger.warning("Failed to resolve proxy exit IP for WebRTC spoofing; removing --fingerprint-webrtc-ip=auto")
|
|
args = list(args)
|
|
del args[idx]
|
|
return args
|
|
if exit_ip:
|
|
args = list(args)
|
|
args[idx] = f"--fingerprint-webrtc-ip={exit_ip}"
|
|
else:
|
|
logger.warning("Could not resolve proxy exit IP for WebRTC spoofing; removing --fingerprint-webrtc-ip=auto")
|
|
args = list(args)
|
|
del args[idx]
|
|
return args
|
|
|
|
|
|
def _append_webrtc_exit_ip(
|
|
args: list[str] | None, exit_ip: str | None
|
|
) -> list[str] | None:
|
|
"""Append ``--fingerprint-webrtc-ip=<exit_ip>`` unless the user already set it.
|
|
|
|
*exit_ip* comes free from the geoip lookup; it spoofs the WebRTC IP to the
|
|
egress IP. No-op when there is no exit IP or the flag is already present. This
|
|
rule must stay identical across every launch path, so it lives in one place.
|
|
"""
|
|
if exit_ip and not (args and any(a.startswith("--fingerprint-webrtc-ip") for a in args)):
|
|
args = list(args or [])
|
|
args.append(f"--fingerprint-webrtc-ip={exit_ip}")
|
|
return args
|
|
|
|
|
|
def build_args(
|
|
stealth_args: bool,
|
|
extra_args: list[str] | None,
|
|
timezone: str | None = None,
|
|
locale: str | None = None,
|
|
headless: bool = True,
|
|
extension_paths: list[str] | None = None,
|
|
start_maximized: bool = False,
|
|
) -> list[str]:
|
|
"""Combine stealth args with user-provided args and locale flags.
|
|
|
|
Deduplicates by flag key (everything before '=').
|
|
Priority: stealth defaults < user args < dedicated params (timezone/locale).
|
|
"""
|
|
seen: dict[str, str] = {}
|
|
|
|
if stealth_args:
|
|
for arg in get_default_stealth_args():
|
|
seen[arg.split("=", 1)[0]] = arg
|
|
|
|
# GPU blocklist bypass:
|
|
# - Headed mode (all platforms): Chromium blocks WebGL on software GPUs
|
|
# in Docker/Xvfb. Flag lets SwiftShader serve WebGL. See issue #56.
|
|
# - Windows (all modes): Chromium's GPU blocklist blocks WebGPU for the
|
|
# Microsoft Basic Render Driver. Dawn's adapter_blocklist bypass alone
|
|
# isn't enough — need this flag too. Linux doesn't need it.
|
|
import platform as _platform
|
|
if not headless or _platform.system() == "Windows":
|
|
seen["--ignore-gpu-blocklist"] = "--ignore-gpu-blocklist"
|
|
|
|
if extra_args:
|
|
for arg in extra_args:
|
|
key = arg.split("=", 1)[0]
|
|
if key in seen:
|
|
logger.debug("Arg override: %s -> %s", seen[key], arg)
|
|
seen[key] = arg
|
|
|
|
# Playwright's default launch args switch off a browser feature that stock Chrome
|
|
# ships enabled. Re-enable it alongside the Windows font-metrics profile so the
|
|
# feature set matches a stock browser rather than a test harness. Merged into any
|
|
# existing --enable-features value rather than added as a second flag.
|
|
if "--fingerprint-windows-font-metrics" in seen:
|
|
key = "--enable-features"
|
|
current = seen[key].split("=", 1)[-1] if key in seen else ""
|
|
features = [f for f in current.split(",") if f]
|
|
if "MediaRouter" not in features:
|
|
features.append("MediaRouter")
|
|
seen[key] = f"{key}={','.join(features)}"
|
|
|
|
# Timezone/locale flags are independent of stealth_args — always inject when set
|
|
if timezone:
|
|
key = "--fingerprint-timezone"
|
|
flag = f"{key}={timezone}"
|
|
if key in seen:
|
|
logger.debug("Arg override: %s -> %s", seen[key], flag)
|
|
seen[key] = flag
|
|
if locale:
|
|
for key in ("--lang", "--fingerprint-locale"):
|
|
flag = f"{key}={locale}"
|
|
if key in seen:
|
|
logger.debug("Arg override: %s -> %s", seen[key], flag)
|
|
seen[key] = flag
|
|
|
|
if extension_paths:
|
|
abs_paths = [os.path.abspath(p) for p in extension_paths]
|
|
ext_val = ",".join(abs_paths)
|
|
|
|
seen["--load-extension"] = f"--load-extension={ext_val}"
|
|
seen["--disable-extensions-except"] = (
|
|
f"--disable-extensions-except={ext_val}"
|
|
)
|
|
|
|
# Open maximized (real Windows Chrome overwhelmingly runs maximized) so the
|
|
# window fills the spoofed screen. Skipped if the caller already chose a
|
|
# window geometry. Gated to binaries where this stays coherent (see
|
|
# binary_supports_maximized_window) — below the gate it would create
|
|
# outerWidth < innerWidth.
|
|
if start_maximized and not any(
|
|
k in seen for k in ("--start-maximized", "--window-size", "--window-position")
|
|
):
|
|
seen["--start-maximized"] = "--start-maximized"
|
|
|
|
return list(seen.values())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Windows-font mismatch warning (Linux only)
|
|
#
|
|
# On Linux the binary spoofs the Windows platform by default, but fonts come
|
|
# from the host OS. A font-less Linux box contradicts the Windows claim and
|
|
# font-fingerprinting anti-bot systems flag the mismatch. Warn once per
|
|
# environment. See docs/chrome40-fpjs-font-minimum-set-investigation.md.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Microsoft-proprietary fonts that signal a real Windows install (absent from
|
|
# ttf-mscorefonts-installer). Keep in sync with issue #395 and
|
|
# docs/chrome40-fpjs-font-minimum-set-investigation.md.
|
|
# Windows OS fonts — ship with Windows itself, so their absence on a
|
|
# Windows-spoofing Linux host degrades results. The two monospace fonts
|
|
# (Consolas + Courier New) are part of the recommended set so the generic
|
|
# `monospace` family resolves to a Windows font. See issue #395.
|
|
_WINDOWS_FONT_TELLS = (
|
|
"Segoe UI",
|
|
"Segoe UI Light",
|
|
"Calibri",
|
|
"Marlett",
|
|
"MS UI Gothic",
|
|
"Franklin Gothic",
|
|
"Consolas",
|
|
"Courier New",
|
|
)
|
|
|
|
# MS Office supplemental fonts, installed as one atomic block by every Office
|
|
# install. Roughly half of real Windows machines have this pack and half do
|
|
# not, so its absence is a perfectly normal Windows setup, NOT a problem —
|
|
# reported as an informational signal only, never a warning.
|
|
_OFFICE_FONT_TELLS = (
|
|
"MT Extra",
|
|
"Century",
|
|
"Century Gothic",
|
|
"MS Reference Specialty",
|
|
"Wingdings 2",
|
|
"Wingdings 3",
|
|
"Book Antiqua",
|
|
"Bookshelf Symbol 7",
|
|
"Monotype Corsiva",
|
|
"Bookman Old Style",
|
|
)
|
|
|
|
# macOS OS fonts — the core set present on essentially every real Mac. Their
|
|
# absence on a macOS-spoofing Linux host degrades results, just as a Windows-
|
|
# spoofing host needs the Windows set above. Several entries (Arial, Georgia,
|
|
# Times New Roman, Courier New, Tahoma) also ship on a Windows-provisioned box,
|
|
# so a partial count on a non-Mac host is normal, not a fault.
|
|
_MAC_FONT_TELLS = (
|
|
"Apple Color Emoji",
|
|
"Arial",
|
|
"Arial Narrow",
|
|
"Arial Unicode MS",
|
|
"Comic Sans MS",
|
|
"Courier",
|
|
"Courier New",
|
|
"Georgia",
|
|
"Gill Sans",
|
|
"Helvetica",
|
|
"Helvetica Neue",
|
|
"Impact",
|
|
"Menlo",
|
|
"Microsoft Sans Serif",
|
|
"Monaco",
|
|
"Tahoma",
|
|
"Times New Roman",
|
|
"Trebuchet MS",
|
|
"Webdings",
|
|
"Wingdings",
|
|
)
|
|
|
|
_font_warning_checked = False
|
|
|
|
|
|
def _count_fonts_present(tells: tuple[str, ...]) -> int | None:
|
|
"""Count how many tell-tale fonts are installed, via fc-list.
|
|
|
|
Returns the number present (0..len(tells)), or None if it can't be
|
|
determined (fc-list missing or errored). Callers must NOT treat None as
|
|
zero — None means "unknown", 0 means "genuinely none installed".
|
|
"""
|
|
import subprocess
|
|
try:
|
|
result = subprocess.run(
|
|
["fc-list"], capture_output=True, text=True, timeout=5
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
if result.returncode != 0:
|
|
return None
|
|
listing = result.stdout.lower()
|
|
return sum(1 for font in tells if font.lower() in listing)
|
|
|
|
|
|
def _windows_fonts_present() -> bool | None:
|
|
"""True if ALL Windows OS fonts are installed, False if any are missing,
|
|
None if unknown. Strict: a partial set is treated as incomplete, since the
|
|
font install is atomic and a missing font degrades the Windows persona.
|
|
"""
|
|
n = _count_fonts_present(_WINDOWS_FONT_TELLS)
|
|
return None if n is None else n == len(_WINDOWS_FONT_TELLS)
|
|
|
|
|
|
def _maybe_warn_windows_fonts(chrome_args: list[str]) -> None:
|
|
"""Warn once when spoofing Windows on a Linux host without the full Windows
|
|
font set.
|
|
|
|
Best-effort and silent on error — never raises. Gated by an in-process flag
|
|
plus a cache-dir marker so it fires at most once per environment. Suppress
|
|
entirely with CLOAKBROWSER_SUPPRESS_FONT_WARNING.
|
|
"""
|
|
global _font_warning_checked
|
|
if _font_warning_checked:
|
|
return
|
|
_font_warning_checked = True
|
|
try:
|
|
import platform
|
|
if os.environ.get("CLOAKBROWSER_SUPPRESS_FONT_WARNING"):
|
|
return
|
|
if platform.system() != "Linux":
|
|
return
|
|
# Effective platform = the last --fingerprint-platform in the final argv
|
|
# (build_args dedups, so there is at most one). None => no Windows spoof.
|
|
effective_platform = None
|
|
for arg in chrome_args:
|
|
if arg.startswith("--fingerprint-platform="):
|
|
effective_platform = arg.split("=", 1)[1].strip().lower()
|
|
if effective_platform != "windows":
|
|
return
|
|
from .config import get_cache_dir
|
|
marker = get_cache_dir() / ".font_warning_shown"
|
|
if marker.exists():
|
|
return
|
|
present = _windows_fonts_present()
|
|
if present is None or present:
|
|
return # full set present, or can't determine — don't warn
|
|
# Write straight to stderr (like the welcome banner and the JS/.NET
|
|
# wrappers) so an app's logging config can't silence it. ASCII-only and
|
|
# swallow-everything: on a legacy Windows console stderr is cp1252/strict
|
|
# and this write is on the launch path.
|
|
try:
|
|
sys.stderr.write(
|
|
"[cloakbrowser] Incomplete Windows font set - installing the full "
|
|
"set is strongly advised for best results when spoofing Windows on "
|
|
"Linux. https://github.com/CloakHQ/cloakbrowser#font-setup-on-linux "
|
|
"(silence: CLOAKBROWSER_SUPPRESS_FONT_WARNING=1)\n"
|
|
)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.write_text("")
|
|
except OSError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _parse_proxy_url(proxy: str) -> dict[str, Any]:
|
|
"""Parse HTTP(S) proxy URL, extracting credentials into separate Playwright fields.
|
|
|
|
Handles: http://user:pass@host:port -> {server: "http://host:port", username: "user", password: "pass"}
|
|
Also handles: no credentials, URL-encoded special chars, missing port,
|
|
and bare proxy strings without a scheme (e.g. 'user:pass@host:port' -> treated as http).
|
|
|
|
SOCKS5 URLs are NOT handled here — they take a dedicated path via
|
|
``_normalize_socks_string_url`` in ``_resolve_proxy_config``.
|
|
"""
|
|
# Bare format: "user:pass@host:port" — urlparse needs a scheme to extract credentials.
|
|
normalized = proxy
|
|
if "@" in proxy and "://" not in proxy:
|
|
normalized = f"http://{proxy}"
|
|
|
|
parsed = urlparse(normalized)
|
|
|
|
if not parsed.username:
|
|
return {"server": proxy} # no creds — return original unchanged
|
|
|
|
# Rebuild server URL without credentials
|
|
netloc = parsed.hostname or ""
|
|
if parsed.port:
|
|
netloc += f":{parsed.port}"
|
|
|
|
server = urlunparse((parsed.scheme, netloc, parsed.path, "", "", ""))
|
|
|
|
result: dict[str, Any] = {"server": server}
|
|
result["username"] = unquote(parsed.username)
|
|
if parsed.password:
|
|
result["password"] = unquote(parsed.password)
|
|
|
|
return result
|
|
|
|
|
|
def _has_credentials(proxy: str | ProxySettings) -> bool:
|
|
"""Check if the proxy has inline or dict-level credentials."""
|
|
if isinstance(proxy, dict):
|
|
return bool(proxy.get("username"))
|
|
return "@" in proxy
|
|
|
|
|
|
def _reconstruct_http_url(proxy: ProxySettings) -> str:
|
|
"""Reconstruct an HTTP(S) proxy URL with inline credentials from a Playwright proxy dict."""
|
|
server = proxy.get("server", "")
|
|
username = proxy.get("username", "")
|
|
password = proxy.get("password", "")
|
|
if not username:
|
|
return server
|
|
parsed = urlparse(_ensure_proxy_scheme(server))
|
|
enc_user = quote(username, safe="")
|
|
enc_pass = quote(password, safe="") if password else None
|
|
return _assemble_proxy_url(
|
|
parsed.scheme, parsed.hostname or "", parsed.port,
|
|
enc_user, enc_pass, parsed.path,
|
|
)
|
|
|
|
|
|
def _normalize_http_string_url(url: str) -> str:
|
|
"""Re-encode credentials in an HTTP(S) proxy URL string for --proxy-server.
|
|
|
|
Same pattern as ``_normalize_socks_string_url`` — decode then re-encode to
|
|
ensure Chromium's proxy URL parser handles special chars correctly.
|
|
"""
|
|
normalized = url if "://" in url else f"http://{url}"
|
|
try:
|
|
parsed = urlparse(normalized)
|
|
_ = parsed.port
|
|
except ValueError as e:
|
|
logger.warning("Malformed HTTP proxy URL, passing through unchanged: %s", e)
|
|
return normalized
|
|
if parsed.username is None and parsed.password is None:
|
|
return normalized
|
|
raw_user = parsed.username or ""
|
|
enc_user = quote(unquote(raw_user), safe="") if raw_user else ""
|
|
if parsed.password is not None:
|
|
raw_pass = parsed.password
|
|
enc_pass = quote(unquote(raw_pass), safe="") if raw_pass else ""
|
|
else:
|
|
raw_pass = None
|
|
enc_pass = None
|
|
result = _assemble_proxy_url(
|
|
parsed.scheme, parsed.hostname or "", parsed.port,
|
|
enc_user, enc_pass,
|
|
parsed.path, parsed.params, parsed.query, parsed.fragment,
|
|
)
|
|
if enc_user != raw_user or enc_pass != raw_pass:
|
|
logger.info(
|
|
"Auto URL-encoded HTTP proxy credentials (special characters "
|
|
"detected). Pre-encode the URL to suppress this notice."
|
|
)
|
|
return result
|
|
|
|
|
|
def _is_socks_proxy(proxy: str | ProxySettings | None) -> bool:
|
|
"""Check if the proxy uses SOCKS5 protocol."""
|
|
if proxy is None:
|
|
return False
|
|
url = proxy.get("server", "") if isinstance(proxy, dict) else proxy
|
|
return url.lower().startswith(("socks5://", "socks5h://"))
|
|
|
|
|
|
_request_proxy_warned = False
|
|
|
|
|
|
def _warn_on_request_client_use(impl_context: Any) -> None:
|
|
"""Warn once, on first use, that Playwright's request client does not use a
|
|
proxy we passed as --proxy-server (#579).
|
|
|
|
Playwright only proxies ``context.request`` when it owns the proxy. Its
|
|
get/post/fetch, ``page.request`` and ``route.fetch()`` all funnel through the
|
|
impl ``APIRequestContext._inner_fetch`` (shared by the sync and async APIs),
|
|
so one instance wrap catches every path.
|
|
"""
|
|
# A per-context proxy does reach the request client, so nothing leaks there.
|
|
if (getattr(impl_context, "_options", None) or {}).get("proxy"):
|
|
return
|
|
request = getattr(impl_context, "_request", None)
|
|
original = getattr(request, "_inner_fetch", None)
|
|
# Private Playwright API: if it is ever renamed, skip rather than break.
|
|
if original is None:
|
|
return
|
|
|
|
async def _inner_fetch(*args: Any, **kwargs: Any) -> Any:
|
|
global _request_proxy_warned
|
|
# Self-remove so only the first call runs through us. pop() because two
|
|
# concurrent first calls can both land here.
|
|
vars(request).pop("_inner_fetch", None)
|
|
if not _request_proxy_warned:
|
|
_request_proxy_warned = True
|
|
# Straight to stderr so an app's logging config can't silence it.
|
|
sys.stderr.write(
|
|
"[cloakbrowser] context.request / page.request / route.fetch() do not use "
|
|
"this proxy (it is set on the browser, not Playwright) and will send from "
|
|
"your real IP. Use page.evaluate('fetch(...)') for proxied requests.\n"
|
|
)
|
|
return await original(*args, **kwargs)
|
|
|
|
request._inner_fetch = _inner_fetch
|
|
|
|
|
|
def _resolve_proxy_config(
|
|
proxy: str | ProxySettings | None,
|
|
browser_version: str | None = None,
|
|
license_key: str | None = None,
|
|
release_channel: str | None = None,
|
|
) -> tuple[dict[str, Any], list[str]]:
|
|
"""Resolve proxy into Playwright kwargs and Chrome args.
|
|
|
|
Proxies with credentials (SOCKS5 always; HTTP/HTTPS only on binaries that
|
|
support inline proxy auth) are passed via Chrome's --proxy-server flag with
|
|
inline credentials, bypassing Playwright's CDP auth interceptor which breaks
|
|
on some proxies and Google domains (#182). HTTP/HTTPS creds on older binaries
|
|
fall back to Playwright's proxy dict.
|
|
|
|
Returns:
|
|
(proxy_kwargs, extra_chrome_args) — one or both will be empty.
|
|
"""
|
|
if proxy is None:
|
|
return {}, []
|
|
|
|
if _is_socks_proxy(proxy):
|
|
# SOCKS5: bypass Playwright, pass directly to Chrome via --proxy-server.
|
|
# Chrome handles SOCKS5 auth natively from the URL.
|
|
if isinstance(proxy, dict):
|
|
url = _reconstruct_socks_url(proxy)
|
|
extra_args = [f"--proxy-server={url}"]
|
|
bypass = proxy.get("bypass")
|
|
if bypass:
|
|
extra_args.append(f"--proxy-bypass-list={bypass}")
|
|
return {}, extra_args
|
|
# String URL — re-encode creds to work around Chromium parser truncating
|
|
# passwords at '=' and other special chars (#157).
|
|
return {}, [f"--proxy-server={_normalize_socks_string_url(proxy)}"]
|
|
|
|
# HTTP/HTTPS with credentials, only on binaries that ship inline proxy auth:
|
|
# use Chrome's native proxy authentication path instead of Playwright's CDP
|
|
# auth interceptor (#182). Older binaries (free macOS/linux-arm64) can't parse
|
|
# inline credentials, so they fall through to the Playwright proxy dict below.
|
|
if _has_credentials(proxy) and binary_supports_http_proxy_inline_auth(license_key, browser_version, release_channel):
|
|
if isinstance(proxy, dict):
|
|
url = _reconstruct_http_url(proxy)
|
|
extra_args = [f"--proxy-server={url}"]
|
|
bypass = proxy.get("bypass")
|
|
if bypass:
|
|
extra_args.append(f"--proxy-bypass-list={bypass}")
|
|
return {}, extra_args
|
|
return {}, [f"--proxy-server={_normalize_http_string_url(proxy)}"]
|
|
|
|
# HTTP/HTTPS without credentials: use Playwright's proxy dict
|
|
if isinstance(proxy, dict):
|
|
return {"proxy": proxy}, []
|
|
return {"proxy": _parse_proxy_url(proxy)}, []
|