Spaces:
Paused
Paused
| import os | |
| import time | |
| import uuid | |
| import asyncio | |
| import logging | |
| from quart import Quart, request, jsonify | |
| from camoufox.async_api import AsyncCamoufox | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='[%(asctime)s] [%(levelname)s] %(message)s', | |
| datefmt='%H:%M:%S', | |
| ) | |
| logger = logging.getLogger("TurnstileSolver") | |
| SECRET_TOKEN = os.environ.get("SOLVER_SECRET", "changeme_please_123") | |
| HTML_TEMPLATE = """ | |
| <!DOCTYPE html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="UTF-8"> | |
| <title>Turnstile Solver</title> | |
| <script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async></script> | |
| </head> | |
| <body> | |
| <!-- cf turnstile --> | |
| </body> | |
| </html> | |
| """ | |
| class TurnstileSolver: | |
| def __init__(self, headless=True): | |
| self.headless = headless | |
| async def solve(self, url, sitekey, action=None, cdata=None, timeout=60): | |
| start = time.time() | |
| browser = None | |
| try: | |
| browser = await AsyncCamoufox(headless=self.headless).start() | |
| page = await browser.new_page() | |
| url_with_slash = url + "/" if not url.endswith("/") else url | |
| turnstile_div = f'<div class="cf-turnstile" data-sitekey="{sitekey}"' | |
| if action: | |
| turnstile_div += f' data-action="{action}"' | |
| if cdata: | |
| turnstile_div += f' data-cdata="{cdata}"' | |
| turnstile_div += '></div>' | |
| page_data = HTML_TEMPLATE.replace("<!-- cf turnstile -->", turnstile_div) | |
| await page.route(url_with_slash, lambda route: route.fulfill(body=page_data, status=200)) | |
| await page.goto(url_with_slash) | |
| token = None | |
| for i in range(int(timeout / 2)): | |
| try: | |
| val = await page.input_value("[name=cf-turnstile-response]", timeout=2000) | |
| if val and len(val) > 20: | |
| token = val | |
| break | |
| else: | |
| try: | |
| await page.click("//div[@class='cf-turnstile']", timeout=1000) | |
| except Exception: | |
| pass | |
| except Exception: | |
| pass | |
| await asyncio.sleep(2) | |
| elapsed = round(time.time() - start, 3) | |
| if token: | |
| return {"status": "success", "value": token, "elapsed_time": elapsed} | |
| else: | |
| return {"status": "failure", "value": None, "elapsed_time": elapsed, "reason": "Timeout"} | |
| except Exception as e: | |
| return {"status": "error", "value": None, "elapsed_time": round(time.time() - start, 3), "reason": str(e)} | |
| finally: | |
| if browser: | |
| try: | |
| await browser.close() | |
| except Exception: | |
| pass | |
| app = Quart(__name__) | |
| results = {} | |
| solver = TurnstileSolver(headless=True) | |
| def check_auth(): | |
| token = request.args.get("token") | |
| return token == SECRET_TOKEN | |
| async def index(): | |
| return jsonify({ | |
| "service": "Turnstile Solver API", | |
| "status": "running", | |
| "endpoints": { | |
| "/turnstile": "GET ?url=...&sitekey=...&token=...", | |
| "/result": "GET ?id=...&token=...", | |
| }, | |
| }) | |
| async def create_task(): | |
| if not check_auth(): | |
| return jsonify({"error": "Unauthorized"}), 401 | |
| url = request.args.get("url") | |
| sitekey = request.args.get("sitekey") | |
| action = request.args.get("action") | |
| cdata = request.args.get("cdata") | |
| if not url or not sitekey: | |
| return jsonify({"error": "url and sitekey required"}), 400 | |
| task_id = str(uuid.uuid4()) | |
| results[task_id] = {"status": "pending"} | |
| async def run_solver(): | |
| try: | |
| result = await solver.solve(url, sitekey, action, cdata) | |
| results[task_id] = result | |
| except Exception as e: | |
| results[task_id] = {"status": "error", "reason": str(e)} | |
| asyncio.create_task(run_solver()) | |
| return jsonify({"task_id": task_id}), 202 | |
| async def get_result(): | |
| if not check_auth(): | |
| return jsonify({"error": "Unauthorized"}), 401 | |
| task_id = request.args.get("id") | |
| if not task_id or task_id not in results: | |
| return jsonify({"error": "invalid task_id"}), 400 | |
| result = results[task_id] | |
| if result.get("status") == "pending": | |
| return jsonify({"status": "pending"}), 202 | |
| elif result.get("status") == "success": | |
| return jsonify(result), 200 | |
| else: | |
| return jsonify(result), 422 |