diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 426ccfd..abc5d39 100644 --- a/grok_register_ttk.py +++ b/grok_register_ttk.py @@ -778,17 +778,59 @@ def apply_headless_to_options(options, headless: bool = True) -> None: # - Google Chrome blocks unpacked extensions; CDP stealth alone often fails CF on servers # - config browser_prefer: auto | chromium | chrome # - config browser_path: absolute override +def _windows_browser_candidates() -> tuple[str, ...]: + """Windows Chrome/Chromium paths (Program Files + common Playwright installs).""" + if sys.platform != "win32": + return () + out: list[str] = [] + local = os.environ.get("LOCALAPPDATA") or "" + pf = os.environ.get("PROGRAMFILES") or r"C:\Program Files" + pf86 = os.environ.get("PROGRAMFILES(X86)") or r"C:\Program Files (x86)" + # Prefer Chromium-like (can load unpacked extensions) over Google Chrome + if local: + # ms-playwright chromium-* / chrome-win64 + try: + pw = os.path.join(local, "ms-playwright") + if os.path.isdir(pw): + for name in sorted(os.listdir(pw), reverse=True): + if not name.lower().startswith("chromium"): + continue + cand = os.path.join(pw, name, "chrome-win64", "chrome.exe") + if os.path.isfile(cand): + out.append(cand) + except Exception: + pass + out.append(os.path.join(local, "Chromium", "Application", "chrome.exe")) + out.append(os.path.join(local, "Google", "Chrome", "Application", "chrome.exe")) + out.append(os.path.join(pf, "Chromium", "Application", "chrome.exe")) + out.append(os.path.join(pf, "Google", "Chrome", "Application", "chrome.exe")) + out.append(os.path.join(pf86, "Google", "Chrome", "Application", "chrome.exe")) + # dedupe preserve order + seen: set[str] = set() + uniq: list[str] = [] + for p in out: + key = p.lower() + if key in seen: + continue + seen.add(key) + uniq.append(p) + return tuple(uniq) + + +_WIN_BROWSERS = _windows_browser_candidates() +# Chromium first (turnstilePatch / --load-extension). On Windows, Playwright +# chromium-* paths contain "chromium" and pass _is_google_chrome_path=False. CHROMIUM_CANDIDATES = ( "/usr/bin/chromium", "/usr/bin/chromium-browser", "/usr/lib/chromium/chromium", "/usr/lib/chromium-browser/chromium-browser", "/snap/bin/chromium", # keep as /snap/bin/chromium — do NOT resolve to /usr/bin/snap -) +) + tuple(p for p in _WIN_BROWSERS if "google" not in p.lower() or "chromium" in p.lower()) CHROME_CANDIDATES = ( "/usr/bin/google-chrome-stable", "/usr/bin/google-chrome", -) +) + tuple(p for p in _WIN_BROWSERS if "google" in p.lower() and "chromium" not in p.lower()) # Legacy flat list (tests / external callers) BROWSER_CANDIDATES = CHROMIUM_CANDIDATES + CHROME_CANDIDATES @@ -1025,6 +1067,18 @@ def create_browser_options(headless=None): options.set_pref("credentials_enable_service", False) except Exception: pass + # Force normal Chrome UA — default headless UA contains "HeadlessChrome" + # which Cloudflare often hard-blocks (Attention Required / blocked page). + ua = (get_user_agent() or "").strip() + if ua: + try: + options.set_user_agent(ua) + except Exception: + try: + options.set_argument(f"--user-agent={ua}") + except Exception: + pass + print(f" [browser] user-agent={ua[:72]}...", flush=True) browser_path = resolve_browser_path() if browser_path: try: @@ -3244,7 +3298,9 @@ function isVisible(node) { const texts = [ 'accept all cookies', 'accept all', 'reject all', 'allow all', 'accept cookies', 'got it', 'i agree', 'agree', - '接受全部', '全部接受', '拒绝全部', '全部拒绝', '同意' + // zh: UI shows "接受所有 Cookie" / "全部拒绝" (not only 接受全部) + '接受所有 cookie', '接受所有', '接受全部', '全部接受', + '拒绝全部', '全部拒绝', '同意' ]; const nodes = Array.from(document.querySelectorAll( 'button, a, [role="button"], [data-testid*="cookie"], [id*="cookie"], [class*="cookie"]' diff --git a/scripts/local_turnstile_probe.py b/scripts/local_turnstile_probe.py new file mode 100644 index 0000000..3ea28ef --- /dev/null +++ b/scripts/local_turnstile_probe.py @@ -0,0 +1,414 @@ +#!/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())