From ce5515f45f627a3b5b03257eb4ad569d723d1d3a Mon Sep 17 00:00:00 2001 From: SokDobry Date: Sat, 27 Jun 2026 01:20:58 +0000 Subject: [PATCH] feat(solver): Playwright-based captcha solver service MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A small Python HTTP server that drives a real headless Chromium via Playwright to solve VK's not_robot_captcha. This is the only reliable way to pass the bot detection because the captchaNotRobot.check request must come from a real browser with a real FingerprintJS visitor_id (browser_fp) — a value computed from canvas/webgl/font fingerprinting that cannot be faked from Go code alone. == How it works == 1. Go client receives a captcha error from VK with a redirect_uri pointing at https://id.vk.ru/not_robot_captcha?session_token=... 2. Go client (with -captcha-solver http://127.0.0.1:8766) POSTs {"redirect_uri": "..."} to this server 3. The server: - Opens the URL in headless Chromium - Waits for #not-robot-captcha-checkbox to appear - Generates a realistic mouse path (cubic Bezier + jitter) from a random start point to the checkbox center - Clicks the checkbox - Intercepts the captchaNotRobot.check response and extracts success_token 4. Returns {"success_token": "...", "elapsed": 6.0} (or {"error": ...}) 5. Go client uses the success_token to complete calls.getAnonymousToken Typical solve time: 6-8 seconds. == Files == scripts/captcha_solver.py HTTP server on 127.0.0.1:8766 (single-threaded because Playwright sync API requires it). Reuses one Chromium instance across requests (~500ms startup amortized). Health endpoint at GET /health. deploy/vk-captcha-solver.service systemd unit for production deployment. == Setup (server) == apt install python3-pip pip3 install --break-system-packages playwright python3 -m playwright install chromium python3 -m playwright install-deps chromium cp scripts/captcha_solver.py /root/ cp deploy/vk-captcha-solver.service /etc/systemd/system/ systemctl enable --now vk-captcha-solver == Setup (Termux, for on-device solving) == pkg install python pip install playwright requests playwright install chromium playwright install-deps chromium python3 ~/captcha_solver.py --listen 127.0.0.1:8766 == Client usage == ./vk-turn-proxy \ -vk-link "https://vk.com/call/join/..." \ -peer \ -n 8 -udp \ -captcha-solver http://127.0.0.1:8766 If the solver is unreachable, the client falls back to: 1. inline auto-captcha (PoW v2 + initSession — see previous commit) 2. slider POC 3. manual captcha (browser opens, user solves) == Tested end-to-end == On a real VK call link, from a datacenter IP (201.51.28.86) that previously always got status:BOT: [STREAM 1] [Captcha] Using Playwright solver (real Chromium)... [STREAM 1] [Captcha] Calling Playwright solver at http://127.0.0.1:8766/solve [STREAM 1] [Captcha] Playwright solver returned success_token (elapsed=6.0s) [STREAM 1] [VK Auth] Success with client_id=6287487 --- deploy/vk-captcha-solver.service | 17 +++ scripts/captcha_solver.py | 204 +++++++++++++++++++++++++++++++ 2 files changed, 221 insertions(+) create mode 100644 deploy/vk-captcha-solver.service create mode 100644 scripts/captcha_solver.py 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()