grok-keygen/scripts/local_turnstile_probe.py

#!/usr/bin/env python3
"""本地 3 组探测：静态解析 / 浏览器+扩展 / x.ai Turnstile 页。

用法（项目根）:
  .venv/Scripts/python.exe scripts/local_turnstile_probe.py
退出码: 0=全过, 1=有失败
"""
from __future__ import annotations

import json
import os
import sys
import time
import traceback
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
os.chdir(ROOT)

# 优先项目 venv 已装 DrissionPage
RESULTS: list[tuple[str, bool, str]] = []


def report(name: str, ok: bool, detail: str) -> None:
    RESULTS.append((name, ok, detail))
    mark = "PASS" if ok else "FAIL"
    print(f"\n=== [{mark}] {name} ===")
    print(detail)


def find_chromium_for_ext() -> str | None:
    """Prefer Chromium that can load unpacked extensions (not Google Chrome)."""
    candidates = [
        os.environ.get("BROWSER_PATH", ""),
        os.environ.get("CHROMIUM_PATH", ""),
        r"C:\Users\Chaos\AppData\Local\ms-playwright\chromium-1228\chrome-win64\chrome.exe",
        r"C:\Users\Chaos\AppData\Local\ms-playwright\chromium-1223\chrome-win64\chrome.exe",
        r"C:\Program Files\Chromium\Application\chrome.exe",
        r"C:\Program Files\Google\Chrome\Application\chrome.exe",
    ]
    for c in candidates:
        c = (c or "").strip()
        if c and Path(c).is_file():
            return c
    return None


def test1_static_and_helpers() -> None:
    """T1: 静态 + 导入 helpers（不启浏览器）"""
    lines: list[str] = []
    try:
        import grok_register_ttk as reg

        reg.load_config()
        lines.append(f"config keys turnstile_max_rounds={reg.config.get('turnstile_max_rounds')}")
        lines.append(f"turnstile_fast_fail_count={reg.config.get('turnstile_fast_fail_count')}")
        lines.append(f"turnstile_retry_after_sec={reg.config.get('turnstile_retry_after_sec')}")

        assert hasattr(reg, "cf_sleep"), "missing cf_sleep"
        assert hasattr(reg, "_read_turnstile_token"), "missing _read_turnstile_token"
        assert hasattr(reg, "_click_turnstile_widget"), "missing _click_turnstile_widget"
        assert hasattr(reg, "getTurnstileToken"), "missing getTurnstileToken"
        assert hasattr(reg, "create_browser_options"), "missing create_browser_options"

        # cf_sleep floor under fast
        reg.configure_perf(fast=True, sleep_scale=0.15)
        t0 = time.time()
        reg.cf_sleep(1.0, floor=0.6)
        elapsed = time.time() - t0
        lines.append(f"cf_sleep(1.0, floor=0.6) under fast took {elapsed:.2f}s")
        if elapsed < 0.55:
            report("T1 static+cf_sleep", False, "\n".join(lines + [f"FAIL: elapsed {elapsed:.2f} < 0.55"]))
            return

        # human_sleep under fast is heavily scaled (expected)
        t0 = time.time()
        reg.human_sleep(1.0)
        hs = time.time() - t0
        lines.append(f"human_sleep(1.0) under fast took {hs:.2f}s (expect <<1)")

        path = reg.resolve_browser_path()
        lines.append(f"resolve_browser_path() -> {path!r}")
        lines.append(f"EXTENSION_PATH exists={Path(reg.EXTENSION_PATH).is_dir()} path={reg.EXTENSION_PATH}")

        # turnstilePatch content
        content = (ROOT / "turnstilePatch" / "content.js").read_text(encoding="utf-8")
        lines.append(f"turnstilePatch content.js size={len(content)}")
        if "tryClickTurnstile" not in content and "firePointer" not in content and "checkbox" not in content.lower():
            lines.append("WARN: content.js may lack click logic")

        report("T1 static+cf_sleep", True, "\n".join(lines))
    except Exception as e:
        report("T1 static+cf_sleep", False, "\n".join(lines + [traceback.format_exc(), str(e)]))


def test2_browser_launch_ext() -> None:
    """T2: 启动浏览器 + 尝试加载 turnstilePatch"""
    lines: list[str] = []
    browser = None
    try:
        import grok_register_ttk as reg
        from DrissionPage import Chromium

        reg.load_config()
        browser_path = find_chromium_for_ext()
        lines.append(f"selected browser={browser_path!r}")
        if not browser_path:
            report("T2 browser+ext", False, "\n".join(lines + ["no browser binary found"]))
            return

        is_google = "Google" in browser_path.replace("\\", "/") and "Chrome" in browser_path
        lines.append(f"looks_like_google_chrome={is_google}")

        # Force path for this probe
        reg.config["browser_path"] = browser_path
        reg.config["browser_prefer"] = "chromium"
        reg.config["browser_headless"] = True  # local CI-like

        opts = reg.create_browser_options(headless=True)
        # ensure path set
        try:
            opts.set_browser_path(browser_path)
        except Exception as e:
            lines.append(f"set_browser_path err: {e}")

        lines.append("launching Chromium(headless)...")
        browser = Chromium(opts)
        page = browser.latest_tab
        page.get("about:blank")
        time.sleep(0.5)
        title = page.title
        lines.append(f"about:blank title={title!r}")

        # Probe webdriver + extension injection surface
        info = page.run_js(
            """
return {
  webdriver: navigator.webdriver,
  userAgent: navigator.userAgent.slice(0, 160),
  chromeRuntime: !!(window.chrome && chrome.runtime),
  hasGrokStealth: !!window.__grokStealthApplied,
};
            """
        )
        lines.append(f"js probe={json.dumps(info, ensure_ascii=False)}")
        ua = str((info or {}).get("userAgent") or "")
        if "HeadlessChrome" in ua:
            lines.append("FAIL: UA still contains HeadlessChrome (create_browser_options should override)")
            report("T2 browser+ext", False, "\n".join(lines))
            return

        # apply CDP stealth and recheck
        reg.apply_stealth_patches(page)
        info2 = page.run_js(
            """
return {
  webdriver: navigator.webdriver,
  hasGrokStealth: !!window.__grokStealthApplied,
  userAgent: navigator.userAgent.slice(0, 160),
};
            """
        )
        lines.append(f"after stealth={json.dumps(info2, ensure_ascii=False)}")

        # If Google Chrome, extension likely blocked — still pass launch if page works
        ok = True
        if is_google:
            lines.append("NOTE: Google Chrome cannot --load-extension; expect no turnstilePatch")
        if not isinstance(info2, dict):
            ok = False
            lines.append("stealth probe failed")
        elif info2.get("webdriver") is True:
            lines.append("WARN: navigator.webdriver still true after stealth")

        report("T2 browser+ext", ok, "\n".join(lines))
    except Exception as e:
        report("T2 browser+ext", False, "\n".join(lines + [traceback.format_exc(), str(e)]))
    finally:
        if browser is not None:
            try:
                browser.quit()
            except Exception:
                pass


def test3_xai_turnstile_page() -> None:
    """T3: 打开 x.ai 注册页，探测 CF/Turnstile 是否出现 + 尝试读 token

    Windows 本地优先 headed（更接近服务器 Xvfb）；无显示再 headless。
    """
    lines: list[str] = []
    browser = None
    try:
        import grok_register_ttk as reg
        from DrissionPage import Chromium

        reg.load_config()
        browser_path = find_chromium_for_ext() or reg.resolve_browser_path()
        lines.append(f"selected browser={browser_path!r}")
        if not browser_path:
            report("T3 x.ai turnstile", False, "no browser")
            return

        # Headless with UA override is more reliable in automation sandboxes.
        # (headed Playwright Chromium can hang on page.get in some hosts)
        use_headless = True
        if os.environ.get("PROBE_HEADED", "").strip() in ("1", "true", "yes"):
            use_headless = False
        reg.config["browser_path"] = browser_path
        reg.config["browser_prefer"] = "chromium"
        reg.config["browser_headless"] = use_headless
        reg.configure_perf(fast=False, sleep_scale=1.0, skip_debug_io=False)
        lines.append(f"mode={'headless' if use_headless else 'headed'}")

        opts = reg.create_browser_options(headless=use_headless)
        try:
            opts.set_browser_path(browser_path)
        except Exception:
            pass

        lines.append("launch + open signup...")
        browser = Chromium(opts)
        page = browser.latest_tab

        url = "https://accounts.x.ai/sign-up?redirect=grok-com"
        page.get(url)
        reg.apply_stealth_patches(page)
        try:
            ua_now = page.run_js("return navigator.userAgent")
            lines.append(f"runtime_ua={(ua_now or '')[:140]}")
        except Exception as e:
            lines.append(f"ua_read_err={e}")

        # wait for load / challenge
        diag_history = []
        email_found = False
        cookie_done = False
        email_btn_done = False
        for i in range(1, 26):
            time.sleep(1)
            if i in (1, 2, 3, 5) and not cookie_done:
                try:
                    okc = reg.dismiss_cookie_banner(page, log_callback=lambda m: lines.append(m))
                    if okc:
                        cookie_done = True
                        lines.append(f"[{i:02d}s] cookie dismissed")
                except Exception as e:
                    lines.append(f"[{i:02d}s] cookie_err={e}")
            if i in (2, 3, 5, 8) and cookie_done and not email_btn_done:
                try:
                    clicked = page.run_js(
                        r"""
const candidates = Array.from(document.querySelectorAll('button, a, [role="button"]'));
const target = candidates.find((node) => {
  const text = (node.innerText || node.textContent || '').replace(/\s+/g, '');
  const lower = text.toLowerCase();
  return (
    text.includes('使用邮箱注册') ||
    lower.includes('signupwithemail') ||
    lower.includes('continuewithemail')
  );
});
if (!target) return false;
target.click();
return true;
                        """
                    )
                    if clicked:
                        email_btn_done = True
                        lines.append(f"[{i:02d}s] clicked 使用邮箱注册")
                        time.sleep(1.5)
                except Exception as e:
                    lines.append(f"[{i:02d}s] email_btn_err={e}")
            try:
                title = page.title or ""
            except Exception:
                title = "?"
            diag = {}
            try:
                diag = page.run_js(
                    """
try {
  const input = document.querySelector('input[name="cf-turnstile-response"], textarea[name="cf-turnstile-response"]');
  const iframes = Array.from(document.querySelectorAll(
    'iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"]'
  ));
  const widgets = Array.from(document.querySelectorAll('div.cf-turnstile, [data-sitekey]'));
  const email = document.querySelector('input[type="email"], input[name="email"], input[autocomplete="email"]');
  return {
    title: document.title,
    url: location.href.slice(0, 120),
    hasEmail: !!email,
    hasInput: !!input,
    tokenLen: input ? String(input.value||'').trim().length : 0,
    iframeCount: iframes.length,
    widgetCount: widgets.length,
    sitekey: widgets[0] ? String(widgets[0].getAttribute('data-sitekey')||'').slice(0,20) : '',
    bodyText: (document.body && document.body.innerText || '').slice(0, 200).replace(/\\s+/g,' '),
    webdriver: !!navigator.webdriver,
  };
} catch(e) { return {err: String(e)}; }
                    """
                )
            except Exception as e:
                diag = {"err": str(e)}
            diag_history.append((i, title, diag))
            if i in (1, 3, 5, 10, 15, 20, 25) or (isinstance(diag, dict) and (diag.get("hasEmail") or diag.get("iframeCount"))):
                lines.append(f"[{i:02d}s] title={title!r} diag={json.dumps(diag, ensure_ascii=False)[:300]}")
            if isinstance(diag, dict) and diag.get("hasEmail"):
                email_found = True
                # stay a few more seconds for CF widget after email form
            if isinstance(diag, dict) and int(diag.get("tokenLen") or 0) >= 80:
                lines.append(f"token ready early at {i}s len={diag.get('tokenLen')}")
                break
            if email_found and i >= 12:
                break

        last = diag_history[-1][2] if diag_history else {}
        lines.append(f"final_diag={json.dumps(last, ensure_ascii=False)}")

        # Try click helpers if CF present
        if isinstance(last, dict) and (last.get("iframeCount") or last.get("hasInput") or last.get("widgetCount")):
            # temporarily set page for helpers that use _get_page
            # monkeypatch: store on module via TabPool style
            try:
                # Many helpers use _get_page(); inject by setting thread-local if TabPool
                from tab_pool import TabPool  # type: ignore
            except Exception:
                pass
            # Direct click via private helpers with page arg pattern
            try:
                click = reg._click_turnstile_widget(page)
                lines.append(f"click_path={click}")
            except Exception as e:
                lines.append(f"click_err={e}")
            time.sleep(3)
            token = ""
            try:
                token = reg._read_turnstile_token(page)
            except Exception as e:
                lines.append(f"read_token_err={e}")
            lines.append(f"token_len_after_click={len(token or '')}")
        else:
            lines.append("no CF widget detected on signup landing (may appear after email step)")

        # Screenshot for debug
        shot_dir = ROOT / "screenshots"
        shot_dir.mkdir(exist_ok=True)
        shot = shot_dir / f"local_probe_{int(time.time())}.png"
        try:
            page.get_screenshot(path=str(shot), full_page=True)
            lines.append(f"screenshot={shot}")
        except Exception as e:
            lines.append(f"screenshot_err={e}")

        # Success criteria for T3:
        # PASS only if we can actually automate signup (email form or interactive CF widget).
        # Hard block page ("you have been blocked") is FAIL — not a solvable Turnstile.
        body = str((last or {}).get("bodyText") or "").lower()
        title = str((last or {}).get("title") or "")
        hard_block = (
            "you have been blocked" in body
            or "unable to access" in body
            or ("attention required" in title.lower() and not (last or {}).get("hasEmail"))
        )
        has_form = bool((last or {}).get("hasEmail"))
        has_cf = bool(
            (last or {}).get("iframeCount")
            or (last or {}).get("widgetCount")
            or (last or {}).get("hasInput")
        )
        if hard_block and not has_form:
            lines.append("FAIL: Cloudflare HARD BLOCK (not interactive Turnstile)")
            ok = False
        elif has_form or has_cf:
            lines.append("page usable for further signup automation")
            if not email_found:
                lines.append("NOTE: email input not found within 25s — may be CF interstitial")
            ok = True
        else:
            lines.append("FAIL criteria: no email form and no CF widget")
            ok = False

        report("T3 x.ai turnstile", ok, "\n".join(lines))
    except Exception as e:
        report("T3 x.ai turnstile", False, "\n".join(lines + [traceback.format_exc(), str(e)]))
    finally:
        if browser is not None:
            try:
                browser.quit()
            except Exception:
                pass


def main() -> int:
    print("local_turnstile_probe — 3 tests")
    print(f"ROOT={ROOT}")
    print(f"python={sys.executable}")
    test1_static_and_helpers()
    test2_browser_launch_ext()
    test3_xai_turnstile_page()
    print("\n======== SUMMARY ========")
    fails = 0
    for name, ok, _ in RESULTS:
        print(f"  {'PASS' if ok else 'FAIL'}  {name}")
        if not ok:
            fails += 1
    print(f"total={len(RESULTS)} fail={fails}")
    return 1 if fails else 0


if __name__ == "__main__":
    raise SystemExit(main())