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