pythonlib: WebGL rotation & leak fixes 0.4.0

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

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}")

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 = {

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

pythonlib/camoufox/webgl/__init__.py

@@ -0,0 +1,3 @@
+from .sample import sample_webgl
+
+__all__ = ['sample_webgl']

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])

pythonlib/camoufox/xpi_dl.py

@@ -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}")

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 <daijro.dev@gmail.com>"]
 license = "MIT"