diff --git a/deploy/vk-captcha-solver.service b/deploy/vk-captcha-solver.service new file mode 100644 index 0000000..3e2e802 --- /dev/null +++ b/deploy/vk-captcha-solver.service @@ -0,0 +1,17 @@ +[Unit] +Description=vk-turn-proxy captcha solver (Playwright + headless Chromium) +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=/usr/bin/python3 /root/captcha_solver.py --listen 127.0.0.1:8766 +Restart=on-failure +RestartSec=5 +StandardOutput=journal +StandardError=journal +TasksMax=256 +LimitNOFILE=4096 + +[Install] +WantedBy=multi-user.target diff --git a/scripts/captcha_solver.py b/scripts/captcha_solver.py new file mode 100644 index 0000000..f91c5cc --- /dev/null +++ b/scripts/captcha_solver.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +""" +vk-turn-proxy captcha solver — local HTTP server that uses Playwright + headless +Chromium to solve VK's not_robot_captcha and return a success_token. + +Run: + pip install playwright requests + playwright install chromium + python3 captcha_solver.py [--listen 127.0.0.1:8766] + +HTTP API: + POST /solve + Body (JSON): {"redirect_uri": "https://id.vk.ru/not_robot_captcha?..."} + Response (JSON): {"success_token": "...", "elapsed": 8.3} + or {"error": "...", "elapsed": 8.3} + +The Go vk-turn-proxy client can be configured (via -captcha-solver flag) to +call this server instead of trying to solve captcha inline. This bypasses +VK's bot detection because the captchaNotRobot.check request comes from a +real Chromium browser with a real FingerprintJS visitor_id. + +Note: Chromium must be installed via `playwright install chromium`. +""" +import argparse +import json +import re +import random +import time +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from playwright.sync_api import sync_playwright + +UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" + +# Reuse a single browser instance across requests (Chromium startup is ~500ms) +_browser_lock = threading.Lock() +_browser = None +_playwright_ctx = None + + +def get_browser(): + global _browser, _playwright_ctx + with _browser_lock: + if _browser is None: + _playwright_ctx = sync_playwright().start() + _browser = _playwright_ctx.chromium.launch( + headless=True, + args=["--no-sandbox", "--disable-blink-features=AutomationControlled"], + ) + return _browser + + +def solve_captcha(redirect_uri: str, timeout: float = 30.0) -> dict: + """Load the captcha page in Chromium, click the checkbox, return success_token.""" + start = time.time() + success_token = {"value": None, "error": None} + + browser = get_browser() + context = browser.new_context( + user_agent=UA, + viewport={"width": 1920, "height": 1080}, + locale="en-US", + extra_http_headers={ + "sec-ch-ua": '"Chromium";v="146", "Not A(Brand";v="24", "Google Chrome";v="146"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Windows"', + }, + ) + context.add_init_script(""" + 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: {}}; + """) + page = context.new_page() + + def on_response(response): + if "captchaNotRobot.check" in response.url: + try: + data = response.json() + if data.get("response", {}).get("success_token"): + success_token["value"] = data["response"]["success_token"] + except Exception as e: + success_token["error"] = f"response parse: {e}" + + page.on("response", on_response) + + try: + page.goto(redirect_uri, wait_until="networkidle", timeout=20000) + checkbox = page.wait_for_selector("#not-robot-captcha-checkbox", timeout=10000) + box = checkbox.bounding_box() + target_x = box["x"] + box["width"] / 2 + target_y = box["y"] + box["height"] / 2 + + # Realistic mouse path (cubic bezier) + start_x = random.randint(100, 400) + start_y = random.randint(100, 400) + steps = 40 + for i in range(steps): + t = i / (steps - 1) + cp1x = start_x + random.uniform(50, 200) + cp1y = start_y + random.uniform(50, 200) + cp2x = target_x - random.uniform(20, 100) + cp2y = target_y - random.uniform(20, 100) + x = (1-t)**3 * start_x + 3*(1-t)**2*t * cp1x + 3*(1-t)*t**2 * cp2x + t**3 * target_x + y = (1-t)**3 * start_y + 3*(1-t)**2*t * cp1y + 3*(1-t)*t**2 * cp2y + t**3 * target_y + page.mouse.move(x, y) + page.wait_for_timeout(20) + + page.wait_for_timeout(300) + page.mouse.click(target_x, target_y) + + # Wait for success_token + deadline = time.time() + timeout + while time.time() < deadline and not success_token["value"] and not success_token["error"]: + page.wait_for_timeout(100) + + except Exception as e: + success_token["error"] = str(e) + finally: + context.close() + + elapsed = time.time() - start + if success_token["value"]: + return {"success_token": success_token["value"], "elapsed": round(elapsed, 1)} + return {"error": success_token["error"] or "no success_token received", "elapsed": round(elapsed, 1)} + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, format, *args): + # Log to stdout with timestamp + print(f"[{time.strftime('%H:%M:%S')}] {args[0]}") + + def do_GET(self): + if self.path == "/health": + self._json(200, {"ok": True, "engine": "playwright"}) + else: + self._json(404, {"error": "not found"}) + + def do_POST(self): + if self.path != "/solve": + self._json(404, {"error": "not found"}) + return + try: + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length).decode("utf-8") + req = json.loads(body) if body else {} + except Exception as e: + self._json(400, {"error": f"bad request: {e}"}) + return + + redirect_uri = req.get("redirect_uri", "") + if not redirect_uri: + self._json(400, {"error": "missing redirect_uri"}) + return + + print(f"[/solve] redirect_uri={redirect_uri[:80]}...") + result = solve_captcha(redirect_uri) + print(f"[/solve] result: token={'OK' if result.get('success_token') else 'FAIL'} elapsed={result.get('elapsed')}s") + self._json(200, result) + + def _json(self, code, obj): + body = json.dumps(obj).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--listen", default="127.0.0.1:8766", help="listen address") + args = ap.parse_args() + + # Warm up the browser instance + print("[init] warming up Chromium...") + get_browser() + print("[init] ready") + + host, port = args.listen.split(":") + # IMPORTANT: use plain HTTPServer (single-threaded), NOT ThreadingHTTPServer. + # Playwright's sync API uses greenlets that are pinned to a single thread — + # calling browser.new_context() from a worker thread raises + # "greenlet.error: Cannot switch to a different thread". Serializing all + # requests on the main thread is fine because captcha solving is rare + # (once per ~10 minutes per stream) and takes ~8s. + from http.server import HTTPServer + srv = HTTPServer((host, int(port)), Handler) + srv.daemon_threads = False + print(f"[listen] http://{args.listen} (single-threaded)") + print("[listen] POST /solve with {\"redirect_uri\": \"...\"} to solve captcha") + try: + srv.serve_forever() + except KeyboardInterrupt: + print("\n[shutdown]") + if _browser: + _browser.close() + if _playwright_ctx: + _playwright_ctx.stop() + + +if __name__ == "__main__": + main()