import random import time import asyncio from typing import Dict, List, Optional from playwright.async_api import Browser, BrowserContext, Page class CloudflareBypass: """Cloudflare bypass implementation based on Kameleo techniques""" # Modern, realistic user agents MODERN_USER_AGENTS = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', ] # Realistic viewport sizes VIEWPORT_SIZES = [ {'width': 1920, 'height': 1080}, {'width': 1366, 'height': 768}, {'width': 1536, 'height': 864}, {'width': 1440, 'height': 900}, {'width': 1280, 'height': 720}, ] # Common languages LANGUAGES = [ 'en-US,en;q=0.9', 'en-GB,en;q=0.9', 'en-CA,en;q=0.9', 'en-AU,en;q=0.9', ] @staticmethod def get_random_user_agent() -> str: """Get a random modern user agent""" return random.choice(CloudflareBypass.MODERN_USER_AGENTS) @staticmethod def get_random_viewport() -> Dict[str, int]: """Get a random realistic viewport size""" return random.choice(CloudflareBypass.VIEWPORT_SIZES) @staticmethod def get_random_language() -> str: """Get a random language preference""" return random.choice(CloudflareBypass.LANGUAGES) @staticmethod def get_stealth_args() -> List[str]: """Get browser arguments for stealth mode""" return [ '--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage', '--disable-accelerated-2d-canvas', '--disable-gpu', '--disable-extensions', '--disable-sync', '--disable-background-networking', '--disable-default-apps', '--disable-translate', '--disable-background-timer-throttling', '--disable-backgrounding-occluded-windows', '--disable-client-side-phishing-detection', '--disable-features=site-per-process', '--disable-hang-monitor', '--disable-ipc-flooding-protection', '--disable-popup-blocking', '--disable-prompt-on-repost', '--disable-renderer-backgrounding', '--memory-pressure-off', '--no-first-run', '--safebrowsing-disable-auto-update', '--max_old_space_size=512', '--disable-web-security', '--disable-features=VizDisplayCompositor', # Additional stealth arguments '--disable-blink-features=AutomationControlled', '--disable-web-security', '--disable-features=VizDisplayCompositor', '--disable-ipc-flooding-protection', '--disable-renderer-backgrounding', '--disable-background-timer-throttling', '--disable-backgrounding-occluded-windows', '--disable-client-side-phishing-detection', '--disable-component-extensions-with-background-pages', '--disable-default-apps', '--disable-domain-reliability', '--disable-features=AudioServiceOutOfProcess', '--disable-hang-monitor', '--disable-prompt-on-repost', '--disable-sync', '--force-color-profile=srgb', '--metrics-recording-only', '--no-first-run', '--password-store=basic', '--use-mock-keychain', '--hide-scrollbars', '--mute-audio', '--no-default-browser-check', '--no-pings', '--no-zygote', '--single-process', '--disable-background-networking', '--disable-background-timer-throttling', '--disable-backgrounding-occluded-windows', '--disable-breakpad', '--disable-component-extensions-with-background-pages', '--disable-dev-shm-usage', '--disable-features=TranslateUI', '--disable-ipc-flooding-protection', '--disable-renderer-backgrounding', '--disable-sync', '--force-color-profile=srgb', '--metrics-recording-only', '--no-first-run', '--safebrowsing-disable-auto-update', '--enable-automation', '--password-store=basic', '--use-mock-keychain', ] @staticmethod async def setup_stealth_context(browser: Browser) -> BrowserContext: """Create a stealth browser context with anti-detection measures""" user_agent = CloudflareBypass.get_random_user_agent() viewport = CloudflareBypass.get_random_viewport() language = CloudflareBypass.get_random_language() # Create context with stealth settings context = await browser.new_context( user_agent=user_agent, viewport=viewport, locale='en-US', timezone_id='America/New_York', permissions=['geolocation'], ignore_https_errors=True, extra_http_headers={ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 'Accept-Language': language, 'Accept-Encoding': 'gzip, deflate, br', 'Accept-Charset': 'utf-8', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', 'Upgrade-Insecure-Requests': '1', } ) # Add stealth scripts to the context await CloudflareBypass._inject_stealth_scripts(context) return context @staticmethod async def _inject_stealth_scripts(context: BrowserContext): """Inject stealth scripts to bypass detection""" await context.add_init_script(""" // Remove webdriver property Object.defineProperty(navigator, 'webdriver', { get: () => undefined, }); // Override permissions const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); // Override plugins Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5], }); // Override languages Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'], }); // Override chrome Object.defineProperty(window, 'chrome', { get: () => ({ runtime: {}, }), }); // Override permissions const originalGetProperty = Object.getOwnPropertyDescriptor; Object.getOwnPropertyDescriptor = function(obj, prop) { if (prop === 'webdriver') { return undefined; } return originalGetProperty(obj, prop); }; // Override toString const originalToString = Function.prototype.toString; Function.prototype.toString = function() { if (this === Function.prototype.toString) { return originalToString.call(this); } if (this === window.navigator.permissions.query) { return 'function query() { [native code] }'; } return originalToString.call(this); }; """) @staticmethod async def setup_stealth_page(page: Page): """Setup stealth measures for a specific page""" # Set additional headers await page.set_extra_http_headers({ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', 'Upgrade-Insecure-Requests': '1', }) # Set realistic timeout page.set_default_timeout(30000) page.set_default_navigation_timeout(30000) # Add human-like behavior await CloudflareBypass._add_human_behavior(page) @staticmethod async def _add_human_behavior(page: Page): """Add human-like behavior to avoid detection""" # Override mouse movement to be more human-like await page.add_init_script(""" // Override mouse events to be more human-like const originalMouseEvent = window.MouseEvent; window.MouseEvent = function(type, init) { if (init && init.movementX === 0 && init.movementY === 0) { init.movementX = Math.random() * 10 - 5; init.movementY = Math.random() * 10 - 5; } return new originalMouseEvent(type, init); }; // Add random mouse movements setInterval(() => { const event = new MouseEvent('mousemove', { clientX: Math.random() * window.innerWidth, clientY: Math.random() * window.innerHeight, movementX: Math.random() * 10 - 5, movementY: Math.random() * 10 - 5, }); document.dispatchEvent(event); }, 5000 + Math.random() * 10000); """) @staticmethod async def handle_cloudflare_challenge(page: Page, max_retries: int = 3) -> bool: """Handle Cloudflare challenges and wait for them to complete""" for attempt in range(max_retries): try: # Check for Cloudflare challenge cloudflare_selectors = [ '#challenge-form', '#cf-please-wait', '.cf-browser-verification', '#cf-wrapper', 'iframe[src*="cloudflare"]' ] for selector in cloudflare_selectors: try: element = await page.wait_for_selector(selector, timeout=5000) if element: print(f"Cloudflare challenge detected on attempt {attempt + 1}") # Wait for challenge to complete await asyncio.sleep(5 + random.uniform(2, 8)) break except: continue # Wait for page to load completely await page.wait_for_load_state('networkidle', timeout=30000) # Check if we're past the challenge title = await page.title() if 'Cloudflare' not in title and 'challenge' not in title.lower(): print("Successfully bypassed Cloudflare challenge") return True # Add some random delay await asyncio.sleep(random.uniform(3, 8)) except Exception as e: print(f"Error handling Cloudflare challenge (attempt {attempt + 1}): {e}") if attempt < max_retries - 1: await asyncio.sleep(random.uniform(5, 15)) return False @staticmethod async def navigate_with_stealth(page: Page, url: str) -> bool: """Navigate to URL with stealth measures and handle Cloudflare""" try: # Add random delay before navigation await asyncio.sleep(random.uniform(1, 3)) # Navigate to the URL await page.goto(url, wait_until='domcontentloaded') # Handle Cloudflare challenge success = await CloudflareBypass.handle_cloudflare_challenge(page) if not success: print("Failed to bypass Cloudflare challenge") return False # Add human-like scrolling await CloudflareBypass._simulate_human_scrolling(page) return True except Exception as e: print(f"Error during stealth navigation: {e}") return False @staticmethod async def _simulate_human_scrolling(page: Page): """Simulate human-like scrolling behavior""" try: # Get page height page_height = await page.evaluate('document.body.scrollHeight') viewport_height = await page.evaluate('window.innerHeight') if page_height > viewport_height: # Scroll down gradually current_position = 0 while current_position < page_height: scroll_amount = random.randint(100, 300) current_position += scroll_amount await page.evaluate(f'window.scrollTo(0, {current_position})') await asyncio.sleep(random.uniform(0.5, 2)) # Scroll back up partially await page.evaluate(f'window.scrollTo(0, {page_height // 3})') await asyncio.sleep(random.uniform(1, 3)) except Exception as e: print(f"Error during human scrolling simulation: {e}") @staticmethod def get_proxy_config(proxy_url: Optional[str] = None) -> Dict[str, str]: """Get proxy configuration if provided""" if not proxy_url: return {} return { 'proxy': { 'server': proxy_url, 'username': '', # Add if needed 'password': '', # Add if needed } }