| """Cloudflare Turnstile solver using Playwright browser automation.
|
|
|
| Supports TurnstileTaskProxyless and TurnstileTaskProxylessM1 task types.
|
| Visits the target page, interacts with the Turnstile widget, and extracts the token.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import asyncio
|
| import logging
|
| from typing import Any
|
|
|
| from playwright.async_api import Browser, Playwright, async_playwright
|
|
|
| from ..core.config import Config
|
|
|
| log = logging.getLogger(__name__)
|
|
|
| _STEALTH_JS = """
|
| Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
|
| Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});
|
| Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
|
| window.chrome = {runtime: {}, loadTimes: () => {}, csi: () => {}};
|
| """
|
|
|
| _EXTRACT_TURNSTILE_TOKEN_JS = """
|
| () => {
|
| // Check for Turnstile response input
|
| const input = document.querySelector('[name="cf-turnstile-response"]')
|
| || document.querySelector('input[name*="turnstile"]');
|
| if (input && input.value && input.value.length > 20) {
|
| return input.value;
|
| }
|
| // Try the turnstile API
|
| if (window.turnstile && typeof window.turnstile.getResponse === 'function') {
|
| const resp = window.turnstile.getResponse();
|
| if (resp && resp.length > 20) return resp;
|
| }
|
| return null;
|
| }
|
| """
|
|
|
|
|
| class TurnstileSolver:
|
| """Solves Cloudflare Turnstile tasks via headless Chromium."""
|
|
|
| def __init__(self, config: Config, browser: Browser | None = None) -> None: |
| self._config = config |
| self._playwright: Playwright | None = None |
| self._browser: Browser | None = browser |
| self._owns_browser = browser is None |
| self._start_lock = asyncio.Lock() |
|
|
| async def start(self) -> None: |
| if self._browser is not None: |
| return |
|
|
| async with self._start_lock: |
| if self._browser is not None: |
| return |
| playwright = await async_playwright().start() |
| try: |
| browser = await playwright.chromium.launch( |
| headless=self._config.browser_headless, |
| args=[ |
| "--disable-blink-features=AutomationControlled", |
| "--no-sandbox", |
| "--disable-dev-shm-usage", |
| "--disable-gpu", |
| ], |
| ) |
| except Exception: |
| await playwright.stop() |
| raise |
| self._playwright = playwright |
| self._browser = browser |
| log.info("TurnstileSolver browser started lazily") |
|
|
| async def stop(self) -> None: |
| async with self._start_lock: |
| if self._owns_browser: |
| if self._browser: |
| await self._browser.close() |
| self._browser = None |
| if self._playwright: |
| await self._playwright.stop() |
| self._playwright = None |
| log.info("TurnstileSolver stopped") |
|
|
| async def solve(self, params: dict[str, Any]) -> dict[str, Any]: |
| await self.start() |
| website_url = params["websiteURL"] |
| website_key = params["websiteKey"] |
|
|
| last_error: Exception | None = None
|
| for attempt in range(self._config.captcha_retries):
|
| try:
|
| token = await self._solve_once(website_url, website_key)
|
| return {"token": token}
|
| except Exception as exc:
|
| last_error = exc
|
| log.warning(
|
| "Turnstile attempt %d/%d failed: %s",
|
| attempt + 1,
|
| self._config.captcha_retries,
|
| exc,
|
| )
|
| if attempt < self._config.captcha_retries - 1:
|
| await asyncio.sleep(2)
|
|
|
| raise RuntimeError(
|
| f"Turnstile failed after {self._config.captcha_retries} attempts: {last_error}"
|
| )
|
|
|
| async def _solve_once(self, website_url: str, website_key: str) -> str:
|
| assert self._browser is not None
|
|
|
| context = await self._browser.new_context(
|
| user_agent=(
|
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
| "AppleWebKit/537.36 (KHTML, like Gecko) "
|
| "Chrome/131.0.0.0 Safari/537.36"
|
| ),
|
| viewport={"width": 1920, "height": 1080},
|
| locale="en-US",
|
| )
|
| page = await context.new_page()
|
| await page.add_init_script(_STEALTH_JS)
|
|
|
| try:
|
| timeout_ms = self._config.browser_timeout * 1000
|
| await page.goto(website_url, wait_until="networkidle", timeout=timeout_ms)
|
|
|
| await page.mouse.move(400, 300)
|
| await asyncio.sleep(1)
|
|
|
|
|
| try:
|
| iframe_element = page.frame_locator(
|
| 'iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"]'
|
| )
|
| checkbox = iframe_element.locator(
|
| 'input[type="checkbox"], .ctp-checkbox-label, label'
|
| )
|
| await checkbox.click(timeout=8_000)
|
| except Exception:
|
| log.info("No Turnstile checkbox found, waiting for auto-solve")
|
|
|
|
|
| for _ in range(15):
|
| await asyncio.sleep(2)
|
| token = await page.evaluate(_EXTRACT_TURNSTILE_TOKEN_JS)
|
| if token:
|
| log.info("Got Turnstile token (len=%d)", len(token))
|
| return token
|
|
|
| raise RuntimeError("Turnstile token not obtained within timeout")
|
| finally:
|
| await context.close()
|
|
|
