646 lines
20 KiB
Python
646 lines
20 KiB
Python
"""
|
|
Stealth Browser Module
|
|
Uses undetected-chromedriver to bypass bot detection (Imperva, Cloudflare, etc.)
|
|
Includes session persistence, proxy rotation, and human-like behavior.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import time
|
|
import random
|
|
import logging
|
|
import pickle
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime, timedelta
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Data directory
|
|
DATA_DIR = Path(__file__).parent.parent / "data"
|
|
|
|
# Session storage directory
|
|
SESSION_DIR = DATA_DIR / "sessions"
|
|
SESSION_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Proxy configuration file
|
|
PROXY_FILE = DATA_DIR / "proxies.json"
|
|
|
|
|
|
class ProxyRotator:
|
|
"""Manages proxy rotation for requests"""
|
|
|
|
def __init__(self, proxy_file: Path = PROXY_FILE):
|
|
self.proxies: List[Dict] = []
|
|
self.current_index = 0
|
|
self.failed_proxies: Dict[str, datetime] = {}
|
|
self.cooldown_minutes = 30
|
|
|
|
self._load_proxies(proxy_file)
|
|
|
|
def _load_proxies(self, proxy_file: Path):
|
|
"""Load proxies from configuration file"""
|
|
if proxy_file.exists():
|
|
try:
|
|
with open(proxy_file, 'r') as f:
|
|
data = json.load(f)
|
|
self.proxies = data.get("proxies", [])
|
|
logger.info(f"Loaded {len(self.proxies)} proxies")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load proxies: {e}")
|
|
|
|
def get_proxy(self) -> Optional[Dict]:
|
|
"""Get next available proxy"""
|
|
if not self.proxies:
|
|
return None
|
|
|
|
# Clean up expired cooldowns
|
|
now = datetime.now()
|
|
self.failed_proxies = {
|
|
k: v for k, v in self.failed_proxies.items()
|
|
if now - v < timedelta(minutes=self.cooldown_minutes)
|
|
}
|
|
|
|
# Find next working proxy
|
|
attempts = 0
|
|
while attempts < len(self.proxies):
|
|
proxy = self.proxies[self.current_index]
|
|
proxy_key = f"{proxy.get('host')}:{proxy.get('port')}"
|
|
|
|
self.current_index = (self.current_index + 1) % len(self.proxies)
|
|
|
|
if proxy_key not in self.failed_proxies:
|
|
return proxy
|
|
|
|
attempts += 1
|
|
|
|
# All proxies in cooldown, return first one anyway
|
|
return self.proxies[0] if self.proxies else None
|
|
|
|
def mark_failed(self, proxy: Dict):
|
|
"""Mark a proxy as failed (temporary cooldown)"""
|
|
if proxy:
|
|
proxy_key = f"{proxy.get('host')}:{proxy.get('port')}"
|
|
self.failed_proxies[proxy_key] = datetime.now()
|
|
logger.warning(f"Proxy {proxy_key} marked as failed")
|
|
|
|
|
|
class StealthBrowser:
|
|
"""
|
|
Stealth browser using undetected-chromedriver.
|
|
Designed to bypass Imperva/Incapsula and similar bot protection.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
headless: bool = False,
|
|
proxy: Optional[Dict] = None,
|
|
user_data_dir: Optional[str] = None,
|
|
session_name: str = "default"
|
|
):
|
|
self.headless = headless
|
|
self.proxy = proxy
|
|
self.user_data_dir = user_data_dir
|
|
self.session_name = session_name
|
|
self.driver = None
|
|
self._setup_complete = False
|
|
|
|
def _get_chrome_options(self):
|
|
"""Configure Chrome options for stealth"""
|
|
import undetected_chromedriver as uc
|
|
|
|
options = uc.ChromeOptions()
|
|
|
|
# Basic stealth settings - compatible with newer Chrome versions
|
|
options.add_argument("--disable-blink-features=AutomationControlled")
|
|
options.add_argument("--disable-dev-shm-usage")
|
|
options.add_argument("--no-sandbox")
|
|
options.add_argument("--disable-infobars")
|
|
|
|
# Headless mode (must be set via options in uc >= 3.5)
|
|
if self.headless:
|
|
options.add_argument("--headless=new")
|
|
|
|
# Window size (realistic resolution)
|
|
options.add_argument("--window-size=1920,1080")
|
|
|
|
# User data directory for session persistence
|
|
if self.user_data_dir:
|
|
options.add_argument(f"--user-data-dir={self.user_data_dir}")
|
|
|
|
# Proxy configuration
|
|
if self.proxy:
|
|
proxy_str = self._format_proxy(self.proxy)
|
|
if proxy_str:
|
|
options.add_argument(f"--proxy-server={proxy_str}")
|
|
|
|
return options
|
|
|
|
def _format_proxy(self, proxy: Dict) -> Optional[str]:
|
|
"""Format proxy dict into Chrome proxy string"""
|
|
if not proxy:
|
|
return None
|
|
|
|
host = proxy.get("host")
|
|
port = proxy.get("port")
|
|
|
|
if not host or not port:
|
|
return None
|
|
|
|
protocol = proxy.get("protocol", "http")
|
|
return f"{protocol}://{host}:{port}"
|
|
|
|
def start(self):
|
|
"""Start the browser"""
|
|
if self.driver:
|
|
return
|
|
|
|
try:
|
|
import undetected_chromedriver as uc
|
|
|
|
options = self._get_chrome_options()
|
|
|
|
# Create driver with version_main to match installed Chrome version
|
|
# This prevents "ChromeDriver only supports Chrome version X" errors
|
|
self.driver = uc.Chrome(
|
|
options=options,
|
|
headless=self.headless,
|
|
use_subprocess=True,
|
|
version_main=146, # Match user's Chrome version
|
|
)
|
|
|
|
# Set realistic viewport
|
|
self.driver.set_window_size(1920, 1080)
|
|
|
|
# Load saved cookies if they exist
|
|
self._load_cookies()
|
|
|
|
self._setup_complete = True
|
|
logger.info("Stealth browser started successfully")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to start stealth browser: {e}")
|
|
raise
|
|
|
|
def stop(self):
|
|
"""Stop the browser and save session"""
|
|
if self.driver:
|
|
try:
|
|
self._save_cookies()
|
|
self.driver.quit()
|
|
except Exception as e:
|
|
logger.warning(f"Error stopping browser: {e}")
|
|
finally:
|
|
self.driver = None
|
|
self._setup_complete = False
|
|
|
|
def is_session_valid(self) -> bool:
|
|
"""
|
|
Check if the browser session is still valid.
|
|
Returns False if the session has crashed or been closed.
|
|
"""
|
|
if not self.driver:
|
|
return False
|
|
|
|
try:
|
|
# Try to get current URL - this will fail if session is invalid
|
|
_ = self.driver.current_url
|
|
return True
|
|
except Exception as e:
|
|
error_msg = str(e).lower()
|
|
dead_session_errors = [
|
|
"invalid session id", "session deleted", "no such session",
|
|
"newconnectionerror", "connection refused", "winerror 10061",
|
|
"failed to establish a new connection", "max retries exceeded",
|
|
"browser has closed",
|
|
]
|
|
if any(err in error_msg for err in dead_session_errors):
|
|
logger.warning(f"Browser session invalid: {e}")
|
|
return False
|
|
# Other errors might be temporary
|
|
return True
|
|
|
|
def restart(self):
|
|
"""
|
|
Restart the browser, cleaning up the old session.
|
|
Useful when the session has become invalid.
|
|
"""
|
|
logger.info("Restarting stealth browser...")
|
|
|
|
# Force cleanup of old driver without trying to save cookies (session is dead)
|
|
if self.driver:
|
|
try:
|
|
self.driver.quit()
|
|
except Exception:
|
|
pass # Ignore errors - session is already dead
|
|
finally:
|
|
self.driver = None
|
|
self._setup_complete = False
|
|
|
|
# Start fresh
|
|
self.start()
|
|
logger.info("Stealth browser restarted successfully")
|
|
|
|
def _get_cookie_file(self) -> Path:
|
|
"""Get path to cookie file for this session"""
|
|
return SESSION_DIR / f"{self.session_name}_cookies.pkl"
|
|
|
|
def _save_cookies(self):
|
|
"""Save cookies to file for session persistence"""
|
|
if not self.driver:
|
|
return
|
|
|
|
try:
|
|
cookies = self.driver.get_cookies()
|
|
cookie_file = self._get_cookie_file()
|
|
|
|
with open(cookie_file, 'wb') as f:
|
|
pickle.dump(cookies, f)
|
|
|
|
logger.debug(f"Saved {len(cookies)} cookies to {cookie_file}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to save cookies: {e}")
|
|
|
|
def _load_cookies(self):
|
|
"""Load cookies from file"""
|
|
cookie_file = self._get_cookie_file()
|
|
|
|
if not cookie_file.exists():
|
|
return
|
|
|
|
try:
|
|
with open(cookie_file, 'rb') as f:
|
|
cookies = pickle.load(f)
|
|
|
|
# Need to visit domain first before adding cookies
|
|
# This will be done when navigating to the actual page
|
|
|
|
self._pending_cookies = cookies
|
|
logger.debug(f"Loaded {len(cookies)} cookies from {cookie_file}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load cookies: {e}")
|
|
self._pending_cookies = []
|
|
|
|
def _apply_pending_cookies(self, domain: str):
|
|
"""Apply loaded cookies after visiting domain"""
|
|
if not hasattr(self, '_pending_cookies') or not self._pending_cookies:
|
|
return
|
|
|
|
for cookie in self._pending_cookies:
|
|
try:
|
|
# Only add cookies for matching domain
|
|
if domain in cookie.get('domain', ''):
|
|
self.driver.add_cookie(cookie)
|
|
except Exception:
|
|
pass # Some cookies may fail, that's ok
|
|
|
|
self._pending_cookies = []
|
|
|
|
def human_delay(self, min_seconds: float = 1.0, max_seconds: float = 3.0):
|
|
"""Add human-like random delay"""
|
|
delay = random.uniform(min_seconds, max_seconds)
|
|
time.sleep(delay)
|
|
|
|
def human_scroll(self):
|
|
"""Scroll like a human would"""
|
|
if not self.driver:
|
|
return
|
|
|
|
# Random scroll amount
|
|
scroll_amount = random.randint(200, 600)
|
|
|
|
# Smooth scroll
|
|
self.driver.execute_script(f"""
|
|
window.scrollBy({{
|
|
top: {scroll_amount},
|
|
behavior: 'smooth'
|
|
}});
|
|
""")
|
|
|
|
self.human_delay(0.5, 1.5)
|
|
|
|
def human_mouse_move(self):
|
|
"""Simulate mouse movement (basic)"""
|
|
if not self.driver:
|
|
return
|
|
|
|
# Move mouse to random position
|
|
try:
|
|
from selenium.webdriver.common.action_chains import ActionChains
|
|
|
|
actions = ActionChains(self.driver)
|
|
|
|
# Random coordinates within viewport
|
|
x = random.randint(100, 800)
|
|
y = random.randint(100, 600)
|
|
|
|
# Move by offset from current position
|
|
actions.move_by_offset(x, y).perform()
|
|
|
|
# Reset position
|
|
actions.move_by_offset(-x, -y).perform()
|
|
except Exception:
|
|
pass # Mouse movement is optional
|
|
|
|
def get_page(self, url: str, wait_time: float = 5.0) -> str:
|
|
"""
|
|
Navigate to URL with human-like behavior.
|
|
|
|
Args:
|
|
url: URL to navigate to
|
|
wait_time: Time to wait for page load
|
|
|
|
Returns:
|
|
Page HTML content
|
|
"""
|
|
if not self.driver:
|
|
self.start()
|
|
|
|
try:
|
|
# Pre-navigation delay
|
|
self.human_delay(0.5, 1.5)
|
|
|
|
# Navigate
|
|
self.driver.get(url)
|
|
|
|
# Apply any pending cookies
|
|
from urllib.parse import urlparse
|
|
domain = urlparse(url).netloc
|
|
self._apply_pending_cookies(domain)
|
|
|
|
# Wait for page load
|
|
time.sleep(wait_time)
|
|
|
|
# Human-like behavior
|
|
self.human_scroll()
|
|
self.human_delay(1, 2)
|
|
self.human_mouse_move()
|
|
|
|
# Get page content
|
|
html = self.driver.page_source
|
|
|
|
# Save cookies after successful page load
|
|
self._save_cookies()
|
|
|
|
return html
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting page {url}: {e}")
|
|
raise
|
|
|
|
def check_for_captcha(self) -> bool:
|
|
"""Check if page has a CAPTCHA challenge blocking content"""
|
|
if not self.driver:
|
|
return False
|
|
|
|
page_source = self.driver.page_source.lower()
|
|
|
|
# First check: Did the page load actual content?
|
|
# If we see product elements, it's NOT a CAPTCHA page
|
|
content_loaded_indicators = [
|
|
'class="product-card',
|
|
'data-product-id',
|
|
'data-sku',
|
|
'/product/',
|
|
'add to cart',
|
|
'product-grid',
|
|
'product-list',
|
|
]
|
|
|
|
for indicator in content_loaded_indicators:
|
|
if indicator in page_source:
|
|
# Page has actual content - no CAPTCHA
|
|
return False
|
|
|
|
# Only flag as CAPTCHA if we see blocking indicators AND no content
|
|
captcha_indicators = [
|
|
"verify you are human",
|
|
"press & hold",
|
|
"press and hold",
|
|
"checking your browser",
|
|
"just a moment",
|
|
"enable javascript and cookies",
|
|
"access denied",
|
|
"blocked",
|
|
"challenge-running",
|
|
"cf-browser-verification",
|
|
"ddos-guard",
|
|
]
|
|
|
|
for indicator in captcha_indicators:
|
|
if indicator in page_source:
|
|
return True
|
|
|
|
# Also check if page is suspiciously empty (might be blocked)
|
|
if len(page_source) < 5000 and "pokemoncenter" not in page_source:
|
|
return True
|
|
|
|
return False
|
|
|
|
def resolve_captcha_interactively(self, url: str, timeout: int = 120) -> bool:
|
|
"""
|
|
When running headless and a CAPTCHA is detected, temporarily pop a visible browser
|
|
window for the user to solve it, then return to headless mode with the updated cookies.
|
|
"""
|
|
if not self.headless:
|
|
# Already visible - just wait in place
|
|
return self.wait_for_captcha_solve(timeout)
|
|
|
|
logger.info("CAPTCHA detected in headless mode - switching to visible browser for manual solve...")
|
|
print("\n" + "=" * 60)
|
|
print("CAPTCHA DETECTED - Opening visible browser window")
|
|
print("Please solve the challenge, then it will return to headless.")
|
|
print("=" * 60 + "\n")
|
|
|
|
# Save current cookies before restarting
|
|
self._save_cookies()
|
|
|
|
# Stop headless driver
|
|
if self.driver:
|
|
try:
|
|
self.driver.quit()
|
|
except Exception:
|
|
pass
|
|
self.driver = None
|
|
self._setup_complete = False
|
|
|
|
# Restart visible
|
|
self.headless = False
|
|
solved = False
|
|
try:
|
|
self.start()
|
|
self.driver.get(url)
|
|
time.sleep(3)
|
|
solved = self.wait_for_captcha_solve(timeout)
|
|
if solved:
|
|
self._save_cookies()
|
|
logger.info("CAPTCHA solved - saving session and returning to headless mode")
|
|
print("\n>>> CAPTCHA solved! Returning to headless mode...\n")
|
|
except Exception as e:
|
|
logger.error(f"Error during interactive CAPTCHA solve: {e}")
|
|
finally:
|
|
# Always return to headless
|
|
if self.driver:
|
|
try:
|
|
self.driver.quit()
|
|
except Exception:
|
|
pass
|
|
self.driver = None
|
|
self._setup_complete = False
|
|
self.headless = True
|
|
try:
|
|
self.start()
|
|
except Exception as e:
|
|
logger.error(f"Failed to restart headless browser after CAPTCHA solve: {e}")
|
|
|
|
return solved
|
|
|
|
def wait_for_captcha_solve(self, timeout: int = 120):
|
|
"""
|
|
Wait for user to solve CAPTCHA manually.
|
|
Only works in non-headless mode.
|
|
"""
|
|
if self.headless:
|
|
logger.warning("Cannot solve CAPTCHA in headless mode")
|
|
return False
|
|
|
|
logger.info("CAPTCHA detected! Please solve it manually...")
|
|
print("\n" + "=" * 50)
|
|
print("CAPTCHA DETECTED!")
|
|
print("Please solve the CAPTCHA in the browser window.")
|
|
print("=" * 50 + "\n")
|
|
|
|
start_time = time.time()
|
|
|
|
while time.time() - start_time < timeout:
|
|
if not self.check_for_captcha():
|
|
logger.info("CAPTCHA solved!")
|
|
self._save_cookies() # Save session after solving
|
|
return True
|
|
|
|
time.sleep(2)
|
|
|
|
logger.warning("CAPTCHA solve timeout")
|
|
return False
|
|
|
|
def screenshot(self, filename: str = "screenshot.png"):
|
|
"""Take a screenshot for debugging"""
|
|
if self.driver:
|
|
try:
|
|
self.driver.save_screenshot(filename)
|
|
logger.info(f"Screenshot saved to {filename}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to save screenshot: {e}")
|
|
|
|
|
|
class StealthBrowserPool:
|
|
"""
|
|
Manages multiple stealth browser instances with proxy rotation.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
pool_size: int = 1,
|
|
use_proxies: bool = False,
|
|
headless: bool = False
|
|
):
|
|
self.pool_size = pool_size
|
|
self.use_proxies = use_proxies
|
|
self.headless = headless
|
|
self.browsers: List[StealthBrowser] = []
|
|
self.proxy_rotator = ProxyRotator() if use_proxies else None
|
|
self.current_index = 0
|
|
|
|
def get_browser(self) -> StealthBrowser:
|
|
"""Get a browser from the pool"""
|
|
# Create browser if pool is empty
|
|
if not self.browsers:
|
|
proxy = self.proxy_rotator.get_proxy() if self.proxy_rotator else None
|
|
browser = StealthBrowser(
|
|
headless=self.headless,
|
|
proxy=proxy,
|
|
session_name=f"pool_{self.current_index}"
|
|
)
|
|
browser.start()
|
|
self.browsers.append(browser)
|
|
return browser
|
|
|
|
# Rotate through browsers
|
|
browser = self.browsers[self.current_index]
|
|
self.current_index = (self.current_index + 1) % len(self.browsers)
|
|
return browser
|
|
|
|
def shutdown_all(self):
|
|
"""Shutdown all browsers in pool"""
|
|
for browser in self.browsers:
|
|
try:
|
|
browser.stop()
|
|
except Exception:
|
|
pass
|
|
self.browsers = []
|
|
|
|
|
|
# Global instances
|
|
_stealth_browser: Optional[StealthBrowser] = None
|
|
_browser_pool: Optional[StealthBrowserPool] = None
|
|
|
|
|
|
def get_stealth_browser(
|
|
headless: bool = False,
|
|
session_name: str = "pokemoncenter"
|
|
) -> StealthBrowser:
|
|
"""Get or create the global stealth browser instance"""
|
|
global _stealth_browser
|
|
|
|
if _stealth_browser is None:
|
|
_stealth_browser = StealthBrowser(
|
|
headless=headless,
|
|
session_name=session_name
|
|
)
|
|
|
|
if not _stealth_browser._setup_complete:
|
|
_stealth_browser.start()
|
|
|
|
return _stealth_browser
|
|
|
|
|
|
def shutdown_stealth_browser():
|
|
"""Shutdown the global stealth browser"""
|
|
global _stealth_browser
|
|
|
|
if _stealth_browser:
|
|
_stealth_browser.stop()
|
|
_stealth_browser = None
|
|
|
|
|
|
def create_proxy_config_template():
|
|
"""Create a template proxies.json file"""
|
|
template = {
|
|
"proxies": [
|
|
{
|
|
"host": "proxy1.example.com",
|
|
"port": 8080,
|
|
"protocol": "http",
|
|
"username": "user",
|
|
"password": "pass"
|
|
},
|
|
{
|
|
"host": "proxy2.example.com",
|
|
"port": 8080,
|
|
"protocol": "http",
|
|
"username": "user",
|
|
"password": "pass"
|
|
}
|
|
],
|
|
"_comment": "Add your residential proxies here. Recommended providers: Bright Data, Oxylabs, Smartproxy"
|
|
}
|
|
|
|
if not PROXY_FILE.exists():
|
|
with open(PROXY_FILE, 'w') as f:
|
|
json.dump(template, f, indent=2)
|
|
logger.info(f"Created proxy template at {PROXY_FILE}")
|
|
|
|
|
|
# Create template on import
|
|
create_proxy_config_template()
|