Create app.py

app.py

@@ -0,0 +1,165 @@
+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
+
+
+@app.route("/")
+async def index():
+    return jsonify({
+        "service": "Turnstile Solver API",
+        "status": "running",
+        "endpoints": {
+            "/turnstile": "GET ?url=...&sitekey=...&token=...",
+            "/result": "GET ?id=...&token=...",
+        },
+    })
+
+
+@app.route("/turnstile")
+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
+
+
+@app.route("/result")
+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
+
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    app.run(host="0.0.0.0", port=port)