Browse Source
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 <host:port> \
-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
pull/183/head
2 changed files with 221 additions and 0 deletions
@ -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 |
||||
@ -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() |
||||
Loading…
Reference in new issue