me/omegafox
1
0
Fork 0
mirror of https://forge.fsky.io/oneflux/omegafox.git synced 2026-02-10 16:42:04 -08:00
Code Issues Projects Releases Packages Wiki Activity Actions

pythonlib: Set window size & kill Xvfb on browser close #49 #50

Browse source 
- Added window: Tuple[int, int] argument to set a fixed width and height #50
- Experimental fix to automatically terminate Xvfb on browser.close() #49
- Bump to 0.3.1
This commit is contained in:
daijro 2024-10-27 20:03:11 -05:00
parent 0040aadc05
commit 75ea7b0880
7 changed files with 222 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
pythonlib/camoufox/async_api.py
View file

@ -1,4 +1,4 @@
from typing import Any, Dict, Optional, Union
from typing import Any, Dict, Optional, Union, overload
from playwright.async_api import (
Browser,
@ -6,8 +6,11 @@ from playwright.async_api import (
Playwright,
PlaywrightContextManager,
)
from typing_extensions import Literal
from .utils import launch_options
from camoufox.virtdisplay import VirtualDisplay
from .utils import async_attach_vd, launch_options
class AsyncCamoufox(PlaywrightContextManager):
@ -32,11 +35,33 @@ class AsyncCamoufox(PlaywrightContextManager):
await super().__aexit__(*args)
@overload
async def AsyncNewBrowser(
playwright: Playwright,
*,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: Literal[False] = False,
**kwargs,
) -> Browser: ...
@overload
async def AsyncNewBrowser(
playwright: Playwright,
*,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: Literal[True],
**kwargs,
) -> BrowserContext: ...
async def AsyncNewBrowser(
playwright: Playwright,
*,
headless: Optional[Union[bool, Literal['virtual']]] = None,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: bool = False,
debug: Optional[bool] = None,
**kwargs,
) -> Union[Browser, BrowserContext]:
"""
@ -50,9 +75,20 @@ async def AsyncNewBrowser(
**kwargs:
All other keyword arugments passed to `launch_options()`.
"""
opt = launch_options(**kwargs)
if headless == 'virtual':
virtual_display = VirtualDisplay(debug=debug)
kwargs['virtual_display'] = virtual_display.get()
headless = False
else:
virtual_display = None
opt = from_options or launch_options(headless=headless, debug=debug, **kwargs)
# Persistent context
if persistent_context:
return await playwright.firefox.launch_persistent_context(**opt)
context = await playwright.firefox.launch_persistent_context(**opt)
return await async_attach_vd(context, virtual_display)
return await playwright.firefox.launch(**opt)
# Browser
browser = await playwright.firefox.launch(**opt)
return await async_attach_vd(browser, virtual_display)

1
pythonlib/camoufox/browserforge.yml
View file

@ -44,6 +44,7 @@ screen:
innerHeight: window.innerHeight
innerWidth: window.innerWidth
screenX: window.screenX
screenY: window.screenY
# Tends to generate out of bounds (network inconsistencies):
# clientWidth: document.body.clientWidth
# clientHeight: document.body.clientHeight

53
pythonlib/camoufox/fingerprints.py
View file

@ -1,9 +1,13 @@
import re
from dataclasses import asdict
from dataclasses import asdict, dataclass
from random import randrange
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, Tuple
from browserforge.fingerprints import Fingerprint, FingerprintGenerator, Screen
from browserforge.fingerprints import (
Fingerprint,
FingerprintGenerator,
ScreenFingerprint,
)
from camoufox.pkgman import load_yaml
@ -13,6 +17,15 @@ BROWSERFORGE_DATA = load_yaml('browserforge.yml')
FP_GENERATOR = FingerprintGenerator(browser='firefox', os=('linux', 'macos', 'windows'))
@dataclass
class ExtendedScreen(ScreenFingerprint):
"""
An extended version of Browserforge's ScreenFingerprint class
"""
screenY: Optional[int] = None
def _cast_to_properties(
camoufox_data: dict, cast_enum: dict, bf_dict: dict, ff_version: Optional[str] = None
) -> None:
@ -40,10 +53,13 @@ def _cast_to_properties(
camoufox_data[type_key] = data
def handle_screenXY(camoufox_data: Dict[str, Any], fp_screen: Screen) -> None:
def handle_screenXY(camoufox_data: Dict[str, Any], fp_screen: ScreenFingerprint) -> None:
"""
Helper method to set window.screenY based on Browserforge's screenX value.
"""
# Skip if manually provided
if 'window.screenY' in camoufox_data:
return
# Default screenX to 0 if not provided
screenX = fp_screen.screenX
if not screenX:
@ -82,10 +98,37 @@ def from_browserforge(fingerprint: Fingerprint, ff_version: Optional[str] = None
return camoufox_data
def generate_fingerprint(**config) -> Fingerprint:
def handle_window_size(fp: Fingerprint, outer_width: int, outer_height: int) -> None:
"""
Helper method to set a custom outer window size, and center it in the screen
"""
# Cast the screen to an ExtendedScreen
fp.screen = ExtendedScreen(**asdict(fp.screen))
sc = fp.screen
# Center the window on the screen
sc.screenX += (sc.width - outer_width) // 2
sc.screenY = (sc.height - outer_height) // 2
# Update inner dimensions if set
if sc.innerWidth:
sc.innerWidth = max(outer_width - sc.outerWidth + sc.innerWidth, 0)
if sc.innerHeight:
sc.innerHeight = max(outer_height - sc.outerHeight + sc.innerHeight, 0)
# Set outer dimensions
sc.outerWidth = outer_width
sc.outerHeight = outer_height
def generate_fingerprint(window: Optional[Tuple[int, int]] = None, **config) -> Fingerprint:
"""
Generates a Firefox fingerprint with Browserforge.
"""
if window: # User-specified outer window size
fingerprint = FP_GENERATOR.generate(**config)
handle_window_size(fingerprint, *window)
return fingerprint
return FP_GENERATOR.generate(**config)

49
pythonlib/camoufox/sync_api.py
View file

@ -1,4 +1,4 @@
from typing import Any, Dict, Optional, Union
from typing import Any, Dict, Optional, Union, overload
from playwright.sync_api import (
Browser,
@ -6,8 +6,11 @@ from playwright.sync_api import (
Playwright,
PlaywrightContextManager,
)
from typing_extensions import Literal
from .utils import launch_options
from camoufox.virtdisplay import VirtualDisplay
from .utils import launch_options, sync_attach_vd
class Camoufox(PlaywrightContextManager):
@ -32,11 +35,33 @@ class Camoufox(PlaywrightContextManager):
super().__exit__(*args)
@overload
def NewBrowser(
playwright: Playwright,
*,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: Literal[False] = False,
**kwargs,
) -> Browser: ...
@overload
def NewBrowser(
playwright: Playwright,
*,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: Literal[True],
**kwargs,
) -> BrowserContext: ...
def NewBrowser(
playwright: Playwright,
*,
headless: Optional[Union[bool, Literal['virtual']]] = None,
from_options: Optional[Dict[str, Any]] = None,
persistent_context: bool = False,
debug: Optional[bool] = None,
**kwargs,
) -> Union[Browser, BrowserContext]:
"""
@ -50,8 +75,20 @@ def NewBrowser(
**kwargs:
All other keyword arugments passed to `launch_options()`.
"""
opt = from_options or launch_options(**kwargs)
if persistent_context:
return playwright.firefox.launch_persistent_context(**opt)
if headless == 'virtual':
virtual_display = VirtualDisplay(debug=debug)
kwargs['virtual_display'] = virtual_display.get()
headless = False
else:
virtual_display = None
return playwright.firefox.launch(**opt)
opt = from_options or launch_options(headless=headless, debug=debug, **kwargs)
# Persistent context
if persistent_context:
context = playwright.firefox.launch_persistent_context(**opt)
return sync_attach_vd(context, virtual_display)
# Browser
browser = playwright.firefox.launch(**opt)
return sync_attach_vd(browser, virtual_display)

67
pythonlib/camoufox/utils.py
View file

@ -34,7 +34,7 @@ from .warnings import LeakWarning
from .xpi_dl import add_default_addons
if OS_NAME == 'lin':
from .virtdisplay import VIRTUAL_DISPLAY
from .virtdisplay import VirtualDisplay
LAUNCH_FILE = {
'win': 'camoufox.exe',
@ -294,6 +294,50 @@ def warn_manual_config(config: Dict[str, Any]) -> None:
LeakWarning.warn('viewport', False)
async def async_attach_vd(
browser: Any, virtual_display: Optional[VirtualDisplay] = None
) -> Any: # type: ignore
"""
Attaches the virtual display to the async browser cleanup
"""
if not virtual_display: # Skip if no virtual display is provided
return browser
_close = browser.close
async def new_close(*args: Any, **kwargs: Any):
await _close(*args, **kwargs)
if virtual_display:
virtual_display.kill()
browser.close = new_close
browser._virtual_display = virtual_display
return browser
def sync_attach_vd(
browser: Any, virtual_display: Optional[VirtualDisplay] = None
) -> Any: # type: ignore
"""
Attaches the virtual display to the sync browser cleanup
"""
if not virtual_display: # Skip if no virtual display is provided
return browser
_close = browser.close
def new_close(*args: Any, **kwargs: Any):
_close(*args, **kwargs)
if virtual_display:
virtual_display.kill()
browser.close = new_close
browser._virtual_display = virtual_display
return browser
def launch_options(
*,
config: Optional[Dict[str, Any]] = None,
@ -308,9 +352,10 @@ def launch_options(
fonts: Optional[List[str]] = None,
exclude_addons: Optional[List[DefaultAddons]] = None,
screen: Optional[Screen] = None,
window: Optional[Tuple[int, int]] = None,
fingerprint: Optional[Fingerprint] = None,
ff_version: Optional[int] = None,
headless: Optional[Union[bool, Literal['virtual']]] = None,
headless: Optional[bool] = None,
executable_path: Optional[str] = None,
firefox_user_prefs: Optional[Dict[str, Any]] = None,
proxy: Optional[Dict[str, str]] = None,
@ -318,6 +363,7 @@ def launch_options(
env: Optional[Dict[str, Union[str, float, bool]]] = None,
i_know_what_im_doing: Optional[bool] = None,
debug: Optional[bool] = None,
virtual_display: Optional[str] = None,
**launch_options: Dict[str, Any],
) -> Dict[str, Any]:
"""
@ -356,6 +402,8 @@ def launch_options(
screen (Optional[Screen]):
Constrains the screen dimensions of the generated fingerprint.
Takes a browserforge.fingerprints.Screen instance.
window (Optional[Tuple[int, int]]):
Set a fixed window size instead of generating a random one
fingerprint (Optional[Fingerprint]):
Use a custom BrowserForge fingerprint. Note: Not all values will be implemented.
If not provided, a random fingerprint will be generated based on the provided
@ -363,9 +411,10 @@ def launch_options(
ff_version (Optional[int]):
Firefox version to use. Defaults to the current Camoufox version.
To prevent leaks, only use this for special cases.
headless (Union[bool, Literal['virtual']]):
headless (Optional[bool]):
Whether to run the browser in headless mode. Defaults to False.
If you are running linux, passing 'virtual' will use Xvfb.
Note: If you are running linux, passing headless='virtual' to Camoufox & AsyncCamoufox
will use Xvfb.
executable_path (Optional[str]):
Custom Camoufox browser executable path.
firefox_user_prefs (Optional[Dict[str, Any]]):
@ -379,6 +428,8 @@ def launch_options(
Environment variables to set.
debug (Optional[bool]):
Prints the config being sent to Camoufox.
virtual_display (Optional[str]):
Virtual display number. Ex: ':99'. This is handled by Camoufox & AsyncCamoufox.
**launch_options (Dict[str, Any]):
Additional Firefox launch options.
"""
@ -402,10 +453,9 @@ def launch_options(
if isinstance(executable_path, str):
executable_path = Path(abspath(executable_path))
# Handle headless mode cases
if headless == 'virtual':
env['DISPLAY'] = VIRTUAL_DISPLAY.new_or_reuse(debug=debug)
headless = False
# Handle virtual display
if virtual_display:
env['DISPLAY'] = virtual_display
# Warn the user for manual config settings
if not i_know_what_im_doing:
@ -433,6 +483,7 @@ def launch_options(
if fingerprint is None:
fingerprint = generate_fingerprint(
screen=screen or get_screen_cons(headless or 'DISPLAY' in env),
window=window,
os=os,
)
else:

47
pythonlib/camoufox/virtdisplay.py
View file

@ -1,6 +1,8 @@
import os
import subprocess # nosec
from glob import glob
from multiprocessing import Lock
from random import randrange
from shutil import which
from typing import List, Optional
@ -17,12 +19,14 @@ class VirtualDisplay:
A minimal virtual display implementation for Linux.
"""
def __init__(self) -> None:
def __init__(self, debug: Optional[bool] = False) -> None:
"""
Constructor for the VirtualDisplay class (singleton object).
"""
self.debug = debug
self.proc: Optional[subprocess.Popen] = None
self._display: Optional[int] = None
self._lock = Lock()
xvfb_args = (
# fmt: off
@ -61,36 +65,46 @@ class VirtualDisplay:
"""
return [self.xvfb_path, f':{self.display}', *self.xvfb_args]
def execute_xvfb_singleton(self, debug: Optional[bool] = False):
def execute_xvfb(self):
"""
Spawn a detatched process
"""
if debug:
if self.debug or True:
print('Starting virtual display:', ' '.join(self.xvfb_cmd))
self.proc = subprocess.Popen( # nosec
self.xvfb_cmd,
stdout=None if debug else subprocess.DEVNULL,
stderr=None if debug else subprocess.DEVNULL,
stdout=None if self.debug else subprocess.DEVNULL,
stderr=None if self.debug else subprocess.DEVNULL,
)
def new_or_reuse(self, debug: Optional[bool] = False) -> str:
def get(self) -> str:
"""
Get the display number
"""
self.assert_linux()
if self.proc is None:
self.execute_xvfb_singleton(debug)
elif debug:
print(f'Using virtual display: {self.display}')
return f':{self.display}'
with self._lock:
if self.proc is None:
self.execute_xvfb()
elif self.debug:
print(f'Using virtual display: {self.display}')
return f':{self.display}'
def __del__(self):
def kill(self):
"""
Terminate the xvfb process
"""
if self.proc:
self.proc.terminate()
with self._lock:
if self.proc and self.proc.poll() is None:
if self.debug:
print('Terminating virtual display:', self.display)
self.proc.terminate()
def __del__(self):
"""
Kill and delete the VirtualDisplay object
"""
self.kill()
@staticmethod
def _get_lock_files() -> List[str]:
@ -112,7 +126,7 @@ class VirtualDisplay:
ls = list(
map(lambda x: int(x.split("X")[1].split("-")[0]), VirtualDisplay._get_lock_files())
)
return max(99, max(ls) + 3) if ls else 99
return max(99, max(ls) + randrange(3, 20)) if ls else 99 # nosec
@property
def display(self) -> int:
@ -130,6 +144,3 @@ class VirtualDisplay:
"""
if OS_NAME != 'lin':
raise VirtualDisplayNotSupported("Virtual display is only supported on Linux.")
VIRTUAL_DISPLAY = VirtualDisplay()

2
pythonlib/pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "camoufox"
version = "0.3.0"
version = "0.3.1"
description = "Wrapper around Playwright to help launch Camoufox"
authors = ["daijro <daijro.dev@gmail.com>"]
license = "MIT"