diff --git a/pythonlib/camoufox/__version__.py b/pythonlib/camoufox/__version__.py index 9f4915f..ff41fb9 100644 --- a/pythonlib/camoufox/__version__.py +++ b/pythonlib/camoufox/__version__.py @@ -8,7 +8,7 @@ class CONSTRAINTS: The minimum and maximum supported versions of the Camoufox browser. """ - MIN_VERSION = 'beta.15' + MIN_VERSION = 'beta.17' MAX_VERSION = '1' @staticmethod diff --git a/pythonlib/camoufox/addons.py b/pythonlib/camoufox/addons.py index 7345cbd..dfff0b6 100644 --- a/pythonlib/camoufox/addons.py +++ b/pythonlib/camoufox/addons.py @@ -1,14 +1,10 @@ -import asyncio import os -import socket -import threading -import time from enum import Enum -from typing import List +from multiprocessing import Lock +from typing import List, Optional -import orjson - -from .exceptions import InvalidAddonPath, InvalidDebugPort, MissingDebugPort +from .exceptions import InvalidAddonPath +from .pkgman import get_path, unzip, webdl class DefaultAddons(Enum): @@ -17,35 +13,6 @@ class DefaultAddons(Enum): """ UBO = "https://addons.mozilla.org/firefox/downloads/latest/ublock-origin/latest.xpi" - # Disable by default. Not always necessary, and increases the memory footprint of Camoufox. - # BPC = "https://gitflic.ru/project/magnolia1234/bpc_uploads/blob/raw?file=bypass_paywalls_clean-latest.xpi" - - -def get_debug_port(args: List[str]) -> int: - """ - Gets the debug port from the args, or creates a new one if not provided - """ - for i, arg in enumerate(args): - # Search for debugger server port - if arg == "-start-debugger-server": - # If arg is found but no port is provided, raise an error - if i + 1 >= len(args): - raise MissingDebugPort(f"No debug port provided: {args}") - debug_port = args[i + 1] - # Try to parse the debug port as an integer - try: - return int(debug_port) - except ValueError as e: - raise InvalidDebugPort( - f"Error parsing debug port. Must be an integer: {debug_port}" - ) from e - - # Create new debugger server port - debug_port_int = get_open_port() - # Add -start-debugger-server {debug_port} to args - args.extend(["-start-debugger-server", str(debug_port_int)]) - - return debug_port_int def confirm_paths(paths: List[str]) -> None: @@ -61,129 +28,62 @@ def confirm_paths(paths: List[str]) -> None: ) -def get_open_port() -> int: +def add_default_addons( + addons_list: List[str], exclude_list: Optional[List[DefaultAddons]] = None +) -> None: """ - Gets an open port + Adds default addons, minus any specified in exclude_list, to addons_list """ - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(('localhost', 0)) - return s.getsockname()[1] + # Build a dictionary from DefaultAddons, excluding keys found in exclude_list + if exclude_list is None: + exclude_list = [] + + addons = [addon for addon in DefaultAddons if addon not in exclude_list] + + with Lock(): + maybe_download_addons(addons, addons_list) -def threaded_try_load_addons(debug_port_int: int, addons_list: List[str]) -> None: +def download_and_extract(url: str, extract_path: str, name: str) -> None: """ - Tries to load addons (in a separate thread) + Downloads and extracts an addon from a given URL to a specified path """ - thread = threading.Thread( - target=try_load_addons, args=(debug_port_int, addons_list), daemon=True - ) - thread.start() + # Create a temporary file to store the downloaded zip + buffer = webdl(url, desc=f"Downloading addon ({name})", bar=False) + unzip(buffer, extract_path, f"Extracting addon ({name})", bar=False) -def try_load_addons(debug_port_int: int, addons_list: List[str]) -> None: +def get_addon_path(addon_name: str) -> str: """ - Tries to load addons + Returns a path to the addon """ - # Wait for the server to be open - while True: + return get_path(os.path.join("addons", addon_name)) + + +def maybe_download_addons( + addons: List[DefaultAddons], addons_list: Optional[List[str]] = None +) -> None: + """ + Downloads and extracts addons from a given dictionary to a specified list + Skips downloading if the addon is already downloaded + """ + for addon in addons: + # Get the addon path + addon_path = get_addon_path(addon.name) + + # Check if the addon is already extracted + if os.path.exists(addon_path): + # Add the existing addon path to addons_list + if addons_list is not None: + addons_list.append(addon_path) + continue + + # Addon doesn't exist, create directory and download try: - with socket.create_connection(("localhost", debug_port_int)): - break - except ConnectionRefusedError: - time.sleep(0.05) - - # Load addons - asyncio.run(load_all_addons(debug_port_int, addons_list)) - - -async def load_all_addons(debug_port_int: int, addons_list: List[str]) -> None: - """ - Loads all addons - """ - addon_loaders = [LoadFirefoxAddon(debug_port_int, addon) for addon in addons_list] - await asyncio.gather(*[addon_loader.load() for addon_loader in addon_loaders]) - - -class LoadFirefoxAddon: - ''' - Firefox addon loader - https://github.com/daijro/hrequests/blob/main/hrequests/extensions.py#L95 - ''' - - def __init__(self, port, addon_path): - self.port: int = port - self.addon_path: str = addon_path - self.success: bool = False - self.buffers: list = [] - self.remaining_bytes: int = 0 - - async def load(self): - reader, writer = await asyncio.open_connection('localhost', self.port) - writer.write(self._format_message({"to": "root", "type": "getRoot"})) - await writer.drain() - - while True: - data = await reader.read(100) # Adjust buffer size as needed - if not data: - break - await self._process_data(writer, data) - - writer.close() - await writer.wait_closed() - return self.success - - async def _process_data(self, writer, data): - while data: - if self.remaining_bytes == 0: - index = data.find(b':') - if index == -1: - self.buffers.append(data) - return - - total_data = b''.join(self.buffers) + data - size, _, remainder = total_data.partition(b':') - - try: - self.remaining_bytes = int(size) - except ValueError as e: - raise ValueError("Invalid state") from e - - data = remainder - - if len(data) < self.remaining_bytes: - self.remaining_bytes -= len(data) - self.buffers.append(data) - return - else: - self.buffers.append(data[: self.remaining_bytes]) - message = orjson.loads(b''.join(self.buffers)) - self.buffers.clear() - - await self._on_message(writer, message) - - data = data[self.remaining_bytes :] - self.remaining_bytes = 0 - - async def _on_message(self, writer, message): - if "addonsActor" in message: - writer.write( - self._format_message( - { - "to": message["addonsActor"], - "type": "installTemporaryAddon", - "addonPath": self.addon_path, - } - ) - ) - await writer.drain() - - if "addon" in message: - self.success = True - writer.write_eof() - - if "error" in message: - writer.write_eof() - - def _format_message(self, data): - raw = orjson.dumps(data) - return f"{len(raw)}:".encode() + raw + os.makedirs(addon_path, exist_ok=True) + download_and_extract(addon.value, addon_path, addon.name) + # Add the new addon directory path to addons_list + if addons_list is not None: + addons_list.append(addon_path) + except Exception as e: + print(f"Failed to download and extract {addon.name}: {e}") diff --git a/pythonlib/camoufox/utils.py b/pythonlib/camoufox/utils.py index e4e3936..517a3ac 100644 --- a/pythonlib/camoufox/utils.py +++ b/pythonlib/camoufox/utils.py @@ -14,12 +14,7 @@ from screeninfo import get_monitors from typing_extensions import TypeAlias from ua_parser import user_agent_parser -from .addons import ( - DefaultAddons, - confirm_paths, - get_debug_port, - threaded_try_load_addons, -) +from .addons import DefaultAddons, add_default_addons, confirm_paths from .exceptions import ( InvalidOS, InvalidPropertyType, @@ -32,7 +27,7 @@ from .locale import geoip_allowed, get_geolocation, handle_locales from .pkgman import OS_NAME, get_path, installed_verstr, launch_path from .virtdisplay import VirtualDisplay from .warnings import LeakWarning -from .xpi_dl import add_default_addons +from .webgl import sample_webgl ListOrString: TypeAlias = Union[Tuple[str, ...], List[str], str] @@ -345,7 +340,8 @@ def launch_options( os: Optional[ListOrString] = None, block_images: Optional[bool] = None, block_webrtc: Optional[bool] = None, - allow_webgl: Optional[bool] = None, + block_webgl: Optional[bool] = None, + webgl_config: Optional[Tuple[str, str]] = None, geoip: Optional[Union[str, bool]] = None, humanize: Optional[Union[bool, float]] = None, locale: Optional[Union[str, List[str]]] = None, @@ -383,7 +379,7 @@ def launch_options( Whether to block all images. block_webrtc (Optional[bool]): Whether to block WebRTC entirely. - allow_webgl (Optional[bool]): + block_webgl (Optional[bool]): Whether to allow WebGL. To prevent leaks, only use this for special cases. geoip (Optional[Union[str, bool]]): Calculate longitude, latitude, timezone, country, & locale based on the IP address. @@ -434,6 +430,8 @@ def launch_options( Prints the config being sent to Camoufox. virtual_display (Optional[str]): Virtual display number. Ex: ':99'. This is handled by Camoufox & AsyncCamoufox. + webgl_config (Optional[Tuple[str, str]]): + Use a specific WebGL vendor/renderer pair. Passed as a tuple of (vendor, renderer). **launch_options (Dict[str, Any]): Additional Firefox launch options. """ @@ -476,6 +474,7 @@ def launch_options( # Confirm all addon paths are valid if addons: confirm_paths(addons) + config['addons'] = addons # Get the Firefox version if ff_version: @@ -512,7 +511,7 @@ def launch_options( config['fonts'] = fonts update_fonts(config, target_os) # Set a fixed font spacing seed - set_into(config, 'fonts:spacing_seed', randint(0, 2147483647)) # nosec + set_into(config, 'fonts:spacing_seed', randint(0, 1_073_741_823)) # nosec # Set geolocation if geoip: @@ -555,29 +554,40 @@ def launch_options( if isinstance(humanize, (int, float)): set_into(config, 'humanize:maxTime', humanize) - # Validate the config - validate_config(config, path=executable_path) + # Set Firefox user preferences + if block_images: + firefox_user_prefs['permissions.default.image'] = 2 + if block_webrtc: + firefox_user_prefs['media.peerconnection.enabled'] = False + if block_webgl: + firefox_user_prefs['webgl.disabled'] = True + else: + # Select a random webgl pair + if webgl_config: + merge_into(config, sample_webgl(target_os, *webgl_config)) + else: + merge_into(config, sample_webgl(target_os)) + # Use software rendering to be less unique + merge_into( + firefox_user_prefs, + { + 'webgl.forbid-software': False, + 'webgl.forbid-hardware': True, + 'webgl.force-enabled': True, + }, + ) + + # Cache previous pages, requests, etc (uses more memory) + if enable_cache: + merge_into(firefox_user_prefs, CACHE_PREFS) # Print the config if debug is enabled if debug: print('[DEBUG] Config:') pprint(config) - # Set Firefox user preferences - if block_images: - firefox_user_prefs['permissions.default.image'] = 2 - if block_webrtc: - firefox_user_prefs['media.peerconnection.enabled'] = False - if allow_webgl: - LeakWarning.warn('allow_webgl', i_know_what_im_doing) - firefox_user_prefs['webgl.disabled'] = False - - # Cache previous pages, requests, etc (uses more memory) - if enable_cache: - firefox_user_prefs.update(CACHE_PREFS) - - # Load the addons - threaded_try_load_addons(get_debug_port(args), addons) + # Validate the config + validate_config(config, path=executable_path) # Prepare environment variables to pass to Camoufox env_vars = { diff --git a/pythonlib/camoufox/warnings.yml b/pythonlib/camoufox/warnings.yml index 4c64f7e..10a87db 100644 --- a/pythonlib/camoufox/warnings.yml +++ b/pythonlib/camoufox/warnings.yml @@ -28,10 +28,6 @@ custom_fingerprint: >- proxy_without_geoip: >- When using a proxy, it is heavily recommended that you pass `geoip=True`. -allow_webgl: >- - Enabling WebGL can lead to Canvas fingerprinting and detection. - Camoufox will automatically spoof your vendor and renderer, but it cannot spoof your WebGL fingerprint. - ff_version: >- Spoofing the Firefox version will likely lead to detection. If rotating the Firefox version is absolutely necessary, it would be more advisable to diff --git a/pythonlib/camoufox/webgl/__init__.py b/pythonlib/camoufox/webgl/__init__.py new file mode 100644 index 0000000..ca911a3 --- /dev/null +++ b/pythonlib/camoufox/webgl/__init__.py @@ -0,0 +1,3 @@ +from .sample import sample_webgl + +__all__ = ['sample_webgl'] diff --git a/pythonlib/camoufox/webgl/sample.py b/pythonlib/camoufox/webgl/sample.py new file mode 100644 index 0000000..9d34e64 --- /dev/null +++ b/pythonlib/camoufox/webgl/sample.py @@ -0,0 +1,82 @@ +import sqlite3 +from pathlib import Path +from typing import Dict, Optional + +import numpy as np +import orjson + + +def sample_webgl( + os: str, vendor: Optional[str] = None, renderer: Optional[str] = None +) -> Dict[str, str]: + """ + Sample a random WebGL vendor/renderer combination and its data based on OS probabilities. + Optionally use a specific vendor/renderer pair. + + Args: + os: Operating system ('win', 'mac', or 'lin') + vendor: Optional specific vendor to use + renderer: Optional specific renderer to use (requires vendor to be set) + + Returns: + Dict containing WebGL data including vendor, renderer and additional parameters + + Raises: + ValueError: If invalid OS provided or no data found for OS/vendor/renderer + """ + # Map OS to probability column + os_map = {'win': 'windows', 'mac': 'macos', 'lin': 'linux'} + if os not in os_map: + raise ValueError(f'Invalid OS: {os}. Must be one of: {", ".join(os_map)}') + os = os_map[os] + + # Get database path relative to this file + db_path = Path(__file__).parent / 'webgl_data.db' + + # Connect to database + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + + if vendor and renderer: + # Get specific vendor/renderer pair and verify it exists for this OS + cursor.execute( + f'SELECT vendor, renderer, data, {os} FROM webgl_fingerprints WHERE vendor = ? AND renderer = ?', + (vendor, renderer), + ) + result = cursor.fetchone() + conn.close() + + if not result: + raise ValueError(f'No WebGL data found for vendor "{vendor}" and renderer "{renderer}"') + + if result[3] <= 0: # Check OS-specific probability + raise ValueError( + f'Vendor "{vendor}" and renderer "{renderer}" combination not valid for {os}' + ) + + return { + 'vendor': result[0], + 'renderer': result[1], + **orjson.loads(result[2]), + } + + # Get all vendor/renderer pairs and their probabilities for this OS + cursor.execute(f'SELECT vendor, renderer, data, {os} FROM webgl_fingerprints WHERE {os} > 0') + results = cursor.fetchall() + conn.close() + + if not results: + raise ValueError(f'No WebGL data found for OS: {os}') + + # Split into separate arrays + _, _, data_strs, probs = map(list, zip(*results)) + + # Convert probabilities to numpy array and normalize + probs_array = np.array(probs, dtype=np.float64) + probs_array = probs_array / probs_array.sum() + + # Sample based on probabilities + idx = np.random.choice(len(probs_array), p=probs_array) + + # Parse the JSON data string + return orjson.loads(data_strs[idx]) diff --git a/pythonlib/camoufox/webgl/webgl_data.db b/pythonlib/camoufox/webgl/webgl_data.db new file mode 100644 index 0000000..e0b06d6 Binary files /dev/null and b/pythonlib/camoufox/webgl/webgl_data.db differ diff --git a/pythonlib/camoufox/xpi_dl.py b/pythonlib/camoufox/xpi_dl.py deleted file mode 100644 index b05822b..0000000 --- a/pythonlib/camoufox/xpi_dl.py +++ /dev/null @@ -1,67 +0,0 @@ -import os -from multiprocessing import Lock -from typing import List, Optional - -from .addons import DefaultAddons -from .pkgman import get_path, unzip, webdl - - -def add_default_addons( - addons_list: List[str], exclude_list: Optional[List[DefaultAddons]] = None -) -> None: - """ - Adds default addons, minus any specified in exclude_list, to addons_list - """ - # Build a dictionary from DefaultAddons, excluding keys found in exclude_list - if exclude_list is None: - exclude_list = [] - - addons = [addon for addon in DefaultAddons if addon not in exclude_list] - - with Lock(): - maybe_download_addons(addons, addons_list) - - -def download_and_extract(url: str, extract_path: str, name: str) -> None: - """ - Downloads and extracts an addon from a given URL to a specified path - """ - # Create a temporary file to store the downloaded zip - buffer = webdl(url, desc=f"Downloading addon ({name})", bar=False) - unzip(buffer, extract_path, f"Extracting addon ({name})", bar=False) - - -def get_addon_path(addon_name: str) -> str: - """ - Returns a path to the addon - """ - return get_path(os.path.join("addons", addon_name)) - - -def maybe_download_addons( - addons: List[DefaultAddons], addons_list: Optional[List[str]] = None -) -> None: - """ - Downloads and extracts addons from a given dictionary to a specified list - Skips downloading if the addon is already downloaded - """ - for addon in addons: - # Get the addon path - addon_path = get_addon_path(addon.name) - - # Check if the addon is already extracted - if os.path.exists(addon_path): - # Add the existing addon path to addons_list - if addons_list is not None: - addons_list.append(addon_path) - continue - - # Addon doesn't exist, create directory and download - try: - os.makedirs(addon_path, exist_ok=True) - download_and_extract(addon.value, addon_path, addon.name) - # Add the new addon directory path to addons_list - if addons_list is not None: - addons_list.append(addon_path) - except Exception as e: - print(f"Failed to download and extract {addon.name}: {e}") diff --git a/pythonlib/pyproject.toml b/pythonlib/pyproject.toml index d24a95b..af7d6b0 100644 --- a/pythonlib/pyproject.toml +++ b/pythonlib/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api" [tool.poetry] name = "camoufox" -version = "0.3.10" +version = "0.4.0" description = "Wrapper around Playwright to help launch Camoufox" authors = ["daijro "] license = "MIT"