409 lines
16 KiB
Python
409 lines
16 KiB
Python
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 = 2) -> bool:
|
|
"""Handle Cloudflare challenge with improved efficiency"""
|
|
for attempt in range(max_retries):
|
|
try:
|
|
print(f"Handling Cloudflare challenge (attempt {attempt + 1}/{max_retries})")
|
|
|
|
# Wait for challenge form with shorter timeout
|
|
try:
|
|
await page.wait_for_selector('#challenge-form', timeout=10000) # 10 seconds
|
|
except:
|
|
# If no challenge form, might already be bypassed
|
|
return True
|
|
|
|
# Wait for the challenge to complete with shorter timeout
|
|
try:
|
|
await page.wait_for_function('''
|
|
() => {
|
|
return !document.querySelector('#challenge-form') &&
|
|
!document.querySelector('#cf-wrapper') &&
|
|
!document.querySelector('#cf-please-wait') &&
|
|
!document.querySelector('.cf-browser-verification');
|
|
}
|
|
''', timeout=15000) # 15 seconds
|
|
|
|
print("Cloudflare challenge completed")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Challenge completion timeout (attempt {attempt + 1}): {e}")
|
|
|
|
# Try to click any "Verify" buttons
|
|
try:
|
|
verify_buttons = await page.query_selector_all('button[type="submit"], input[type="submit"]')
|
|
for button in verify_buttons:
|
|
try:
|
|
await button.click(timeout=5000)
|
|
await asyncio.sleep(2)
|
|
except:
|
|
continue
|
|
except:
|
|
pass
|
|
|
|
# Short delay before retry
|
|
if attempt < max_retries - 1:
|
|
await asyncio.sleep(2)
|
|
|
|
except Exception as e:
|
|
print(f"Error handling Cloudflare challenge (attempt {attempt + 1}): {e}")
|
|
if attempt < max_retries - 1:
|
|
await asyncio.sleep(2)
|
|
|
|
print("Failed to bypass Cloudflare challenge after all attempts")
|
|
return False
|
|
|
|
@staticmethod
|
|
async def navigate_with_stealth(page: Page, url: str) -> bool:
|
|
"""Navigate to URL with stealth measures and improved performance"""
|
|
try:
|
|
# Set shorter timeouts for better performance
|
|
page.set_default_timeout(30000) # 30 seconds
|
|
page.set_default_navigation_timeout(30000) # 30 seconds
|
|
|
|
# Navigate with stealth measures
|
|
response = await page.goto(url, wait_until='domcontentloaded') # Changed from 'networkidle' to 'domcontentloaded' for speed
|
|
|
|
if not response:
|
|
return False
|
|
|
|
# Quick check for Cloudflare challenge
|
|
if response.status == 403 or response.status == 503:
|
|
# Handle Cloudflare challenge
|
|
return await CloudflareBypass.handle_cloudflare_challenge(page, max_retries=2) # Reduced retries
|
|
|
|
# Check for Cloudflare indicators in the page
|
|
cloudflare_detected = await page.evaluate('''() => {
|
|
return document.title.toLowerCase().includes('cloudflare') ||
|
|
!!document.querySelector('#challenge-form') ||
|
|
!!document.querySelector('#cf-wrapper') ||
|
|
!!document.querySelector('#cf-please-wait') ||
|
|
!!document.querySelector('.cf-browser-verification');
|
|
}''')
|
|
|
|
if cloudflare_detected:
|
|
return await CloudflareBypass.handle_cloudflare_challenge(page, max_retries=2) # Reduced retries
|
|
|
|
# Wait for page to be ready (shorter wait)
|
|
try:
|
|
await page.wait_for_load_state('domcontentloaded', timeout=10000) # 10 second timeout
|
|
except:
|
|
pass # Continue even if load state timeout
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error during stealth navigation to {url}: {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
|
|
}
|
|
} |