Compare commits

...
2 Commits
Author SHA1 Message Date
chaos 9e0f45ac3b Replace CF path with zip register solver; keep Gmail verification
Use grok-register.zip getTurnstileToken/fill_profile and pure MV2
turnstilePatch. Browser options match ReinerBRO headed flags.
Local gmail worker + IMAP + defaultDomains catch-all stay intact.
2026-07-12 07:09:45 +08:00
chaos 60e24b89c1 Align Turnstile with zip refs and harden interactive CF path
Restore MV2 turnstilePatch (content.js), zip-style headed browser flags,
system-default browser path, and interactive-safe solver that skips reset
while the human checkbox is up to avoid 验证失败 after the button appears.
2026-07-12 07:04:38 +08:00
6 changed files with 665 additions and 578 deletions

No files matched your search

+5 -4
View File
@@ -10,7 +10,7 @@
"proxy": "", "proxy": "",
"enable_nsfw": true, "enable_nsfw": true,
"register_count": 1, "register_count": 1,
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36",
"grok2api_auto_add_local": false, "grok2api_auto_add_local": false,
"grok2api_local_token_file": "", "grok2api_local_token_file": "",
"grok2api_pool_name": "ssoBasic", "grok2api_pool_name": "ssoBasic",
@@ -102,7 +102,8 @@
"turnstile_retry_after_sec": 10, "turnstile_retry_after_sec": 10,
"turnstile_retry_gap_sec": 8, "turnstile_retry_gap_sec": 8,
"turnstile_max_rounds": 20, "turnstile_max_rounds": 20,
"browser_prefer": "chrome", "browser_prefer": "auto",
"browser_path": "/usr/bin/google-chrome-stable", "browser_path": "",
"browser_headless": false "browser_headless": false,
"stealth_mode": "off"
} }
+193 -541
View File
@@ -613,23 +613,25 @@ def add_token_to_grok2api_pools(raw_token, email="", log_callback=None):
_add_token_to_grok2api_pools_sync(raw_token, email=email, log_callback=log_callback) _add_token_to_grok2api_pools_sync(raw_token, email=email, log_callback=log_callback)
# Base flags always safe. Keep close to Git-creat7/grokRegister-cpa which only # Headed flags from grok-register.zip (ReinerBRO): no --disable-images, no AutomationControlled.
# uses auto_port + extension (no slim flags) on real headed desktops. # Images must stay ON or Turnstile iframe often fails to paint.
CHROMIUM_BASE_FLAGS = [ CHROMIUM_BASE_FLAGS = [
"--no-sandbox",
"--disable-dev-shm-usage",
# Keep images ON: Turnstile/CF widgets often need them; was a common fail cause.
"--mute-audio",
"--no-first-run", "--no-first-run",
"--disable-blink-features=AutomationControlled", "--no-default-browser-check",
"--disable-background-networking",
"--disable-component-update",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
] ]
# Extra flags only for true headless (no DISPLAY). Under Xvfb/headed these can # Headless / Linux-server extras only. --disable-software-rasterizer blanks CF on headed.
# blank CF widgets (iframe=0 / token=0) — especially --disable-software-rasterizer.
CHROMIUM_HEADLESS_EXTRA_FLAGS = [ CHROMIUM_HEADLESS_EXTRA_FLAGS = [
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu", "--disable-gpu",
"--disable-software-rasterizer", "--disable-software-rasterizer",
"--disable-background-networking", "--mute-audio",
] ]
# Back-compat alias used by older call sites / register_cli patches # Back-compat alias used by older call sites / register_cli patches
@@ -999,18 +1001,10 @@ def _first_existing_browser(candidates) -> str | None:
def resolve_browser_path(): def resolve_browser_path():
"""Pick Chromium/Chrome for registration. """Zip style: only return path if config/env override exists.
Priority: browser_prefer=auto/system → None (DrissionPage system Chrome).
1. config browser_path (absolute)
2. env BROWSER_PATH / CHROME_PATH / CHROMIUM_PATH
3. browser_prefer=chromium|chrome|auto
auto → Chromium first when turnstilePatch exists (needed for CF)
4. any remaining candidate
Never returns /usr/bin/snap (common snap symlink collapse bug).
""" """
# Optional override from config.json "browser_path"
override = str((config.get("browser_path") if isinstance(config, dict) else "") or "").strip() override = str((config.get("browser_path") if isinstance(config, dict) else "") or "").strip()
if override: if override:
norm = _normalize_browser_path(override) norm = _normalize_browser_path(override)
@@ -1018,7 +1012,7 @@ def resolve_browser_path():
return norm return norm
print(f" [browser] ignore invalid browser_path={override!r}", flush=True) print(f" [browser] ignore invalid browser_path={override!r}", flush=True)
for env_key in ("BROWSER_PATH", "CHROMIUM_PATH", "CHROME_PATH"): for env_key in ("BROWSER_PATH", "CHROME_PATH", "CHROMIUM_PATH"):
env_path = (os.environ.get(env_key) or "").strip() env_path = (os.environ.get(env_key) or "").strip()
if not env_path: if not env_path:
continue continue
@@ -1028,28 +1022,22 @@ def resolve_browser_path():
print(f" [browser] ignore invalid {env_key}={env_path!r}", flush=True) print(f" [browser] ignore invalid {env_key}={env_path!r}", flush=True)
prefer = str((config.get("browser_prefer") if isinstance(config, dict) else "") or "auto").strip().lower() prefer = str((config.get("browser_prefer") if isinstance(config, dict) else "") or "auto").strip().lower()
if prefer in ("", "default", "detect"): if prefer in ("", "default", "detect", "system"):
prefer = "auto" prefer = "auto"
# env override
env_pref = (os.environ.get("BROWSER_PREFER") or "").strip().lower() env_pref = (os.environ.get("BROWSER_PREFER") or "").strip().lower()
if env_pref: if env_pref:
prefer = env_pref prefer = env_pref
has_ext = os.path.isdir(EXTENSION_PATH) if prefer in ("auto", "system"):
return None
if prefer in ("chrome", "google-chrome", "google"): if prefer in ("chrome", "google-chrome", "google"):
order = list(CHROME_CANDIDATES) + list(CHROMIUM_CANDIDATES) order = list(CHROME_CANDIDATES) + list(CHROMIUM_CANDIDATES)
elif prefer in ("chromium", "chromeium"): # typo tolerant elif prefer in ("chromium", "chromeium"):
order = list(CHROMIUM_CANDIDATES) + list(CHROME_CANDIDATES) order = list(CHROMIUM_CANDIDATES) + list(CHROME_CANDIDATES)
else: else:
# auto: extension needs Chromium; without extension Chrome is fine return None
if has_ext:
order = list(CHROMIUM_CANDIDATES) + list(CHROME_CANDIDATES)
else:
order = list(CHROME_CANDIDATES) + list(CHROMIUM_CANDIDATES)
return _first_existing_browser(order) return _first_existing_browser(order)
def apply_stealth_patches(page=None, log_callback=None) -> bool: def apply_stealth_patches(page=None, log_callback=None) -> bool:
"""Minimal anti-automation patch via CDP (Chrome cannot load unpacked extensions). """Minimal anti-automation patch via CDP (Chrome cannot load unpacked extensions).
@@ -1057,8 +1045,9 @@ def apply_stealth_patches(page=None, log_callback=None) -> bool:
previously produced xAI [permission_denied] HTTP 403 on signup APIs. previously produced xAI [permission_denied] HTTP 403 on signup APIs.
config.stealth_mode: off | minimal(default) | full config.stealth_mode: off | minimal(default) | full
""" """
mode = str((config.get("stealth_mode") if isinstance(config, dict) else "") or "minimal").strip().lower() # zip 默认不注入 CDP stealth(靠扩展);未配置时也走 off,避免和 content.js 叠指纹
if mode in ("0", "false", "off", "none", "disabled"): mode = str((config.get("stealth_mode") if isinstance(config, dict) else "") or "off").strip().lower()
if mode in ("0", "false", "off", "none", "disabled", ""):
return False return False
page = page if page is not None else _get_page() page = page if page is not None else _get_page()
if page is None: if page is None:
@@ -1097,73 +1086,73 @@ def apply_stealth_patches(page=None, log_callback=None) -> bool:
def create_browser_options(headless=None): def create_browser_options(headless=None):
"""Build ChromiumOptions for register / TabPool. """Zip-style Chromium options (grok-register / ReinerBRO).
headless: Critical for CF:
None → resolve from config/env/auto (no DISPLAY → headless) - always load turnstilePatch
True → force headless - no --disable-images
False → prefer headed (still headless if no DISPLAY) - no browser_path unless config/env set
- no AutomationControlled / heavy stealth flags on headed
""" """
options = ChromiumOptions() options = ChromiumOptions()
options.auto_port() options.auto_port()
options.set_timeouts(base=1) options.set_timeouts(base=1)
# ReinerBRO headed flags
for flag in (
"--no-first-run",
"--no-default-browser-check",
"--disable-background-networking",
"--disable-component-update",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
):
options.set_argument(flag)
# Decide headless first so we can choose flag set (headed ≈ reference repo). use_headless = should_use_headless(headless, log_hint=True)
use_hl = should_use_headless(headless, log_hint=True) if use_headless:
min_browser = str(os.environ.get("GROK_MIN_BROWSER", "") or "").strip().lower() in ( apply_headless_to_options(options, True)
"1", "true", "yes", "on", for flag in (
) "--no-sandbox",
if min_browser: "--disable-dev-shm-usage",
print(" [browser] minimal flags mode (Git-creat7 style)", flush=True) "--disable-gpu",
options.set_argument("--disable-blink-features=AutomationControlled") "--mute-audio",
options.set_argument("--no-first-run") ):
if use_hl:
options.set_argument("--disable-gpu")
else:
for flag in CHROMIUM_BASE_FLAGS:
options.set_argument(flag) options.set_argument(flag)
if use_hl: print(" [browser] headless=true", flush=True)
for flag in CHROMIUM_HEADLESS_EXTRA_FLAGS: else:
options.set_argument(flag) print(" [browser] headless=false (zip headed)", flush=True)
# Do NOT pass --excludeSwitches as a CLI arg (invalid); only real prefs.
try: # system default unless override
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() browser_path = resolve_browser_path()
if browser_path: if browser_path:
try: try:
options.set_browser_path(browser_path) options.set_browser_path(browser_path)
print(f" [browser] path={browser_path}", flush=True) print(f" [browser] path={browser_path}", flush=True)
except Exception: except Exception as exc:
pass print(f" [browser] set path failed: {exc}", flush=True)
else:
print(" [browser] path=system-default (zip style)", flush=True)
# Force UA from config (Chrome/138 etc.)
try:
ua = get_user_agent()
if ua:
options.set_user_agent(ua)
print(f" [browser] ua={ua[:60]}...", flush=True)
except Exception:
pass
if os.path.exists(EXTENSION_PATH): if os.path.exists(EXTENSION_PATH):
# Google Chrome blocks CLI unpacked extension loading; Chromium still allows it. try:
if _is_google_chrome_path(browser_path):
print(
" [ext] Chrome: minimal CDP stealth only (extension blocked)\n"
" 建议安装 Chromium 或 config browser_prefer=chromium / browser_path=/usr/bin/chromium\n"
" 否则 Turnstile token 常为 0(无头/Xvfb 服务器上尤其明显)",
flush=True,
)
else:
options.add_extension(EXTENSION_PATH) options.add_extension(EXTENSION_PATH)
print(f" [ext] turnstilePatch loaded: {EXTENSION_PATH}", flush=True) print(f" [ext] turnstilePatch loaded: {EXTENSION_PATH}", flush=True)
# Apply config.json "proxy" to Chromium. Without this, only HTTP helpers except Exception as ext_exc:
# used get_proxies(); the browser itself fell through to system/env proxy. print(f" [ext] turnstilePatch load failed: {ext_exc}", flush=True)
proxy = (config.get("proxy") or "").strip() else:
print(f" [ext] turnstilePatch missing: {EXTENSION_PATH}", flush=True)
proxy = (config.get("proxy") or "").strip() if isinstance(config, dict) else ""
if proxy: if proxy:
try: try:
from urllib.parse import urlparse from urllib.parse import urlparse
@@ -1173,23 +1162,12 @@ def create_browser_options(headless=None):
if host: if host:
port = u.port or (443 if (u.scheme or "http") == "https" else 80) port = u.port or (443 if (u.scheme or "http") == "https" else 80)
scheme = u.scheme or "http" scheme = u.scheme or "http"
# Chromium --proxy-server cannot embed user:pass
options.set_argument(f"--proxy-server={scheme}://{host}:{port}") options.set_argument(f"--proxy-server={scheme}://{host}:{port}")
print(f" [proxy] browser --proxy-server={scheme}://{host}:{port}", flush=True)
except Exception as e: except Exception as e:
print(f" [proxy] set browser proxy failed: {e}") print(f" [proxy] set browser proxy failed: {e}", flush=True)
apply_headless_to_options(options, use_hl)
if use_hl:
print(" [browser] headless=True", flush=True)
else:
print(
f" [browser] headed DISPLAY={os.environ.get('DISPLAY', '')!r} "
f"(flags=base-only, no disable-software-rasterizer)",
flush=True,
)
return options return options
def _build_request_kwargs(**kwargs): def _build_request_kwargs(**kwargs):
request_kwargs = dict(kwargs) request_kwargs = dict(kwargs)
proxies = request_kwargs.pop("proxies", None) proxies = request_kwargs.pop("proxies", None)
@@ -4154,283 +4132,10 @@ return 'clicked';
raise Exception("验证码已获取,但自动填写/提交失败") raise Exception("验证码已获取,但自动填写/提交失败")
def _read_turnstile_token(page):
"""Read cf-turnstile-response / turnstile.getResponse if present.
Prefer DrissionPage native attr read (sees closed-shadow-ish values better),
then JS walk of open shadow roots + turnstile.getResponse().
"""
# Path 1: native element value (historical success tokens were length ~773)
for loc in (
"@name=cf-turnstile-response",
"css:input[name='cf-turnstile-response']",
"css:textarea[name='cf-turnstile-response']",
"tag:input@@name=cf-turnstile-response",
):
try:
el = page.ele(loc, timeout=0.15)
except Exception:
el = None
if el is None:
continue
try:
val = el.attr("value")
except Exception:
val = None
if not val:
try:
val = el.value
except Exception:
val = None
if not val:
try:
val = page.run_js("return arguments[0] ? String(arguments[0].value||'') : '';", el)
except Exception:
val = None
val = str(val or "").strip()
if len(val) >= 80:
return val
# Path 2: JS light DOM + open shadow + API
try:
token = page.run_js(
"""
try {
function walk(root, acc, depth) {
if (!root || depth > 6) return acc;
try {
const nodes = root.querySelectorAll
? root.querySelectorAll('input[name="cf-turnstile-response"], textarea[name="cf-turnstile-response"]')
: [];
for (const n of nodes) acc.push(n);
} catch (e) {}
let children = [];
try { children = root.querySelectorAll ? root.querySelectorAll('*') : []; } catch (e) { children = []; }
for (const el of children) {
try {
if (el.shadowRoot) walk(el.shadowRoot, acc, depth + 1);
} catch (e) {}
}
return acc;
}
const inputs = walk(document, [], 0);
for (const input of inputs) {
const v = String((input && input.value) || '').trim();
if (v) return v;
}
if (window.turnstile && typeof turnstile.getResponse === 'function') {
try {
const r = String(turnstile.getResponse() || '').trim();
if (r) return r;
} catch (e) {}
try {
const r2 = String(turnstile.getResponse('') || '').trim();
if (r2) return r2;
} catch (e) {}
}
return '';
} catch(e) { return ''; }
"""
)
return str(token or "").strip()
except Exception:
return ""
def _turnstile_diag(page) -> dict:
"""Lightweight CF widget diagnostics for logs (no secrets).
Searches light DOM + open shadow roots. Detects managed-mode "Verifying..."
spinner text so callers can avoid reset/click storms.
"""
try:
info = page.run_js(
"""
try {
function walk(root, acc, depth) {
if (!root || depth > 6) return acc;
try {
const q = root.querySelectorAll
? root.querySelectorAll(
'input[name="cf-turnstile-response"], textarea[name="cf-turnstile-response"], ' +
'iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"], ' +
'div.cf-turnstile, [data-sitekey], iframe[title*="Cloudflare"], iframe[title*="cloudflare"], iframe[title*="Widget"]'
)
: [];
for (const n of q) acc.push(n);
} catch (e) {}
let children = [];
try { children = root.querySelectorAll ? root.querySelectorAll('*') : []; } catch (e) { children = []; }
for (const el of children) {
try { if (el.shadowRoot) walk(el.shadowRoot, acc, depth + 1); } catch (e) {}
}
return acc;
}
const all = walk(document, [], 0);
const inputs = all.filter((n) => n && (n.tagName === 'INPUT' || n.tagName === 'TEXTAREA'));
const iframes = all.filter((n) => n && n.tagName === 'IFRAME');
const widgets = all.filter((n) => n && n.tagName !== 'IFRAME' && n.tagName !== 'INPUT' && n.tagName !== 'TEXTAREA');
const input = inputs[0] || document.querySelector('input[name="cf-turnstile-response"], textarea[name="cf-turnstile-response"]');
const srcs = iframes.slice(0, 3).map((f) => {
try { return String(f.src || f.getAttribute('src') || '').slice(0, 120); } catch(e) { return ''; }
});
const bodyText = ((document.body && document.body.innerText) || '').slice(0, 4000);
const verifying = /正在验证|verifying|checking|just a moment|请完成|人机验证/i.test(bodyText);
// success checkmark / aria
const successish = /成功|success|verified/i.test(bodyText) && !!input && String(input.value||'').trim().length >= 80;
let sitekey = '';
try {
const w = widgets[0] || document.querySelector('[data-sitekey], div.cf-turnstile');
if (w) sitekey = String(w.getAttribute('data-sitekey') || '').slice(0, 16);
} catch (e) {}
// iframe box size (0x0 means not painted)
let iframeBox = '';
try {
if (iframes[0]) {
const r = iframes[0].getBoundingClientRect();
iframeBox = Math.round(r.width) + 'x' + Math.round(r.height);
}
} catch (e) {}
return {
hasInput: !!input,
tokenLen: input ? String(input.value || '').trim().length : 0,
iframeCount: iframes.length,
widgetCount: widgets.length,
sitekey: sitekey,
iframeSrc0: srcs[0] || '',
iframeBox: iframeBox,
turnstileApi: !!(window.turnstile && typeof turnstile.getResponse === 'function'),
webdriver: !!navigator.webdriver,
verifying: verifying,
successish: successish,
};
} catch(e) {
return { err: String(e && e.message || e) };
}
"""
)
return info if isinstance(info, dict) else {"raw": str(info)[:200]}
except Exception as exc:
return {"err": str(exc)[:160]}
def _click_turnstile_widget(page, log_callback=None) -> str:
"""Best-effort click path for interactive Turnstile checkbox.
Ported from ReinerBRO/grok-register + Git-creat7/grokRegister-cpa:
input[name=cf-turnstile-response] → parent shadow → iframe
→ patch screenX/Y inside iframe → body.shadow_root → input.click()
Returns a short tag describing which path succeeded (or 'none').
"""
# Path A: classic ReinerBRO shadow-root checkbox
try:
challenge_input = page.ele("@name=cf-turnstile-response", timeout=0.6)
except Exception:
challenge_input = None
if challenge_input:
iframe = None
try:
wrapper = challenge_input.parent()
iframe = wrapper.shadow_root.ele("tag:iframe")
except Exception:
iframe = None
if iframe is None:
try:
iframe = page.ele(
'css:iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"]',
timeout=0.6,
)
except Exception:
iframe = None
if iframe:
# Match ReinerBRO exactly: value: not get: (CF reads own props).
try:
iframe.run_js(
"""
window.dtp = 1;
function getRandomInt(min, max) {
return Math.floor(Math.random() * (max - min + 1)) + min;
}
let screenX = getRandomInt(800, 1200);
let screenY = getRandomInt(400, 600);
Object.defineProperty(MouseEvent.prototype, 'screenX', { value: screenX });
Object.defineProperty(MouseEvent.prototype, 'screenY', { value: screenY });
try {
if (typeof PointerEvent !== 'undefined' && PointerEvent.prototype) {
Object.defineProperty(PointerEvent.prototype, 'screenX', { value: screenX });
Object.defineProperty(PointerEvent.prototype, 'screenY', { value: screenY });
}
} catch (e) {}
"""
)
except Exception:
pass
# A1: classic ReinerBRO path — body.shadow_root → tag:input → click
try:
body_sr = iframe.ele("tag:body").shadow_root
btn = body_sr.ele("tag:input") or body_sr.ele("css:.mark") or body_sr.ele("css:label")
if btn:
try:
btn.click()
except Exception:
try:
btn.click(by_js=True)
except Exception:
page.actions.click(btn)
return "shadow-checkbox"
except Exception:
pass
# A2: click iframe element itself (center)
try:
iframe.click()
return "iframe-click"
except Exception:
pass
try:
page.actions.click(iframe)
return "iframe-actions"
except Exception:
pass
# Path B: DOM query + synthetic pointer events (Git-creat7 fallback)
try:
clicked = page.run_js(
"""
const nodes = Array.from(document.querySelectorAll(
'iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"], div.cf-turnstile, [data-sitekey]'
)).filter((n) => {
const txt = (n.className || '') + ' ' + (n.id || '') + ' ' + (n.getAttribute?.('src') || '');
return String(txt).toLowerCase().includes('turnstile')
|| String(n.getAttribute?.('src') || '').includes('challenges.cloudflare.com')
|| n.hasAttribute?.('data-sitekey');
});
let ok = false;
for (const n of nodes) {
try {
if (typeof n.click === 'function') { n.click(); ok = true; }
} catch (e) {}
}
return ok;
"""
)
if clicked:
return "js-click"
except Exception:
pass
return "none"
def getTurnstileToken(log_callback=None, cancel_callback=None, max_rounds: int = 20): def getTurnstileToken(log_callback=None, cancel_callback=None, max_rounds: int = 20):
"""batch100-era solver (55f56d2/c56d0d2) with longer waits.
Historical success: Google Chrome + CDP stealth, Accept All Cookies,
reset + shadow checkbox click, token ~773.
"""
page = _get_page() page = _get_page()
if page is None: if page is None:
raise Exception("页面未就绪,无法执行 Turnstile") raise Exception("页面未就绪,无法执行 Turnstile")
apply_stealth_patches(page, log_callback=None)
try: try:
page.run_js( page.run_js(
@@ -4439,20 +4144,10 @@ def getTurnstileToken(log_callback=None, cancel_callback=None, max_rounds: int =
except Exception: except Exception:
pass pass
rounds = max(8, int(max_rounds or 20)) rounds = max(1, int(max_rounds or 20))
# Initial settle for widget attach for _ in range(0, rounds):
cf_sleep(1.2 if not PERF_FLAGS.get("fast") else 0.9, cancel_callback, floor=0.7)
for round_i in range(0, rounds):
raise_if_cancelled(cancel_callback) raise_if_cancelled(cancel_callback)
try: try:
token = _read_turnstile_token(page)
if len(token) >= 80:
if log_callback:
log_callback(f"[*] Turnstile 已通过,token长度={len(token)}")
return token
# Historical pure JS read path as well
token = page.run_js( token = page.run_js(
""" """
try { try {
@@ -4468,21 +4163,55 @@ try {
token = str(token or "").strip() token = str(token or "").strip()
if len(token) >= 80: if len(token) >= 80:
if log_callback: if log_callback:
log_callback(f"[*] Turnstile 已通过,token长度={len(token)}") log_callback(f"[Debug] Turnstile 已通过,token长度={len(token)}")
return token return token
path = _click_turnstile_widget(page, log_callback=log_callback) challenge_input = page.ele("@name=cf-turnstile-response")
if log_callback and (round_i == 0 or round_i % 3 == 0): if challenge_input:
log_callback(f"[Debug] Turnstile click path={path} round={round_i+1}/{rounds}") wrapper = challenge_input.parent()
iframe = None
try:
iframe = wrapper.shadow_root.ele("tag:iframe")
except Exception:
iframe = None
if iframe:
try:
iframe.run_js(
"""
window.dtp = 1;
function getRandomInt(min, max) { return Math.floor(Math.random() * (max - min + 1)) + min; }
let sx = getRandomInt(800, 1200);
let sy = getRandomInt(400, 700);
Object.defineProperty(MouseEvent.prototype, 'screenX', { value: sx });
Object.defineProperty(MouseEvent.prototype, 'screenY', { value: sy });
"""
)
except Exception:
pass
try:
body_sr = iframe.ele("tag:body").shadow_root
btn = body_sr.ele("tag:input")
if btn:
btn.click()
except Exception:
pass
else:
# 兜底:尝试触发页面上可见的 Turnstile 容器
page.run_js(
"""
const nodes = Array.from(document.querySelectorAll('div,span,iframe')).filter((n) => {
const txt = (n.className || '') + ' ' + (n.id || '') + ' ' + (n.getAttribute?.('src') || '');
return String(txt).toLowerCase().includes('turnstile');
});
if (nodes.length && typeof nodes[0].click === 'function') nodes[0].click();
"""
)
except Exception: except Exception:
pass pass
# Progressive wait like c56d0d2, but wall-clock via cf_sleep sleep_with_cancel(1, cancel_callback)
cf_sleep(0.9 + 0.15 * min(round_i, 6), cancel_callback, floor=0.8)
raise Exception("Turnstile 获取 token 失败") raise Exception("Turnstile 获取 token 失败")
def build_profile(): def build_profile():
given_name_pool = [ given_name_pool = [
"Neo", "Ethan", "Liam", "Noah", "Lucas", "Mason", "Ryan", "Leo", "Neo", "Ethan", "Liam", "Noah", "Lucas", "Mason", "Ryan", "Leo",
@@ -4509,112 +4238,18 @@ def build_profile():
def fill_profile_and_submit(timeout=120, log_callback=None, cancel_callback=None): def fill_profile_and_submit(timeout=120, log_callback=None, cancel_callback=None):
"""Fill profile form and submit after Turnstile is solved.
CF wait / retry loop follows Git-creat7/grokRegister-cpa:
fill → if wait-cloudflare → after ~12s call getTurnstileToken (reset+click)
→ re-fill token into hidden input → submit.
"""
page = _get_page() page = _get_page()
apply_stealth_patches(page, log_callback=log_callback)
check_timeout(time.time()) check_timeout(time.time())
# Cookie overlay often reappears on profile step and blanks Turnstile (iframe=0).
dismiss_cookie_banner(page, log_callback=log_callback)
dump_state(page, "profile-form") dump_state(page, "profile-form")
take_screenshot(page, "profile-form") take_screenshot(page, "profile-form")
given_name, family_name, password = build_profile() given_name, family_name, password = build_profile()
# 预热 Turnstile:cookie 只关一次(sticky),然后给 Verifying 时间
if log_callback: if log_callback:
log_callback("[*] 预热 Turnstile...") log_callback("[*] 预热 Turnstile...")
try: human_sleep(2, cancel_callback)
dismiss_cookie_banner(page, log_callback=log_callback)
except Exception:
pass
# 必须用 cf_sleep:--fast 下 human_sleep 会把预热压到 <0.3s,iframe 来不及挂载
cf_sleep(2.4 if not PERF_FLAGS.get("fast") else 1.6, cancel_callback, floor=1.2)
try:
pre = _read_turnstile_token(page)
if pre and len(pre) >= 80 and log_callback:
log_callback(f"[*] 预热阶段已有 token,长度={len(pre)}")
else:
diag_pre = _turnstile_diag(page)
if log_callback:
log_callback(
f"[Debug] 预热后 Turnstile: tokenLen={diag_pre.get('tokenLen', 0)} "
f"hasInput={diag_pre.get('hasInput')} api={diag_pre.get('turnstileApi')} "
f"iframe={diag_pre.get('iframeCount', 0)}"
)
except Exception:
pass
deadline = time.time() + timeout deadline = time.time() + timeout
form_filled_once = False form_filled_once = False
wait_cf_since = None wait_cf_since = None
last_cf_retry_at = 0.0 last_cf_retry_at = 0.0
last_cf_log_at = 0.0
cf_token_fail_streak = 0
try:
# Git-creat7 风格:卡住就二次复用;允许多轮,别太早放弃
max_cf_token_fails = max(3, int(config.get("turnstile_fast_fail_count", 3) or 3))
except Exception:
max_cf_token_fails = 3
try:
# 参考实现:等 12s 后开始 getTurnstileToken,间隔 8s
cf_retry_after = float(config.get("turnstile_retry_after_sec", 45) or 45)
except Exception:
cf_retry_after = 12.0
try:
cf_retry_gap = float(config.get("turnstile_retry_gap_sec", 8) or 8)
except Exception:
cf_retry_gap = 8.0
try:
# 参考 15-20 轮 reset+click;默认抬到 16
turnstile_rounds = max(12, int(config.get("turnstile_max_rounds", 16) or 16))
except Exception:
turnstile_rounds = 16
def _sync_cf_token(token: str):
return page.run_js(
"""
const token = String(arguments[0] || '').trim();
const cfInput = document.querySelector('input[name="cf-turnstile-response"]');
if (!cfInput || !token) return false;
const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set;
if (nativeSetter) nativeSetter.call(cfInput, token);
else cfInput.value = token;
cfInput.dispatchEvent(new Event('input', { bubbles: true }));
cfInput.dispatchEvent(new Event('change', { bubbles: true }));
return String(cfInput.value || '').trim().length;
""",
token,
)
def _retry_turnstile(reason: str) -> bool:
nonlocal cf_token_fail_streak, last_cf_retry_at
if log_callback:
log_callback(f"[*] {reason}")
try:
token = getTurnstileToken(
log_callback=log_callback,
cancel_callback=cancel_callback,
max_rounds=turnstile_rounds,
)
if token:
cf_token_fail_streak = 0
synced = _sync_cf_token(token)
if log_callback:
log_callback(f"[*] Turnstile 二次复用完成,回填长度={synced}")
last_cf_retry_at = time.time()
return True
except Exception as cf_exc:
cf_token_fail_streak += 1
if log_callback:
log_callback(
f"[Debug] Turnstile 二次复用失败({cf_token_fail_streak}/{max_cf_token_fails}): {cf_exc}"
)
if cf_token_fail_streak >= max_cf_token_fails:
raise Exception(f"Turnstile 连续失败,快速放弃资料页: {cf_exc}")
last_cf_retry_at = time.time()
return False
while time.time() < deadline: while time.time() < deadline:
raise_if_cancelled(cancel_callback) raise_if_cancelled(cancel_callback)
@@ -4672,7 +4307,7 @@ const buttons = Array.from(document.querySelectorAll('button[type="submit"], but
}); });
const submitBtn = buttons.find((node) => { const submitBtn = buttons.find((node) => {
const t = (node.innerText || node.textContent || '').replace(/\\s+/g, '').toLowerCase(); const t = (node.innerText || node.textContent || '').replace(/\\s+/g, '').toLowerCase();
return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount') || t.includes('completesignup'); return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount');
}); });
// 必须等待 Cloudflare 校验通过后再提交 // 必须等待 Cloudflare 校验通过后再提交
@@ -4697,39 +4332,44 @@ return 'filled-no-submit';
if isinstance(filled, str) and filled.startswith("wait-cloudflare"): if isinstance(filled, str) and filled.startswith("wait-cloudflare"):
form_filled_once = True form_filled_once = True
if log_callback:
token_len = filled.split(":", 1)[1] if ":" in filled else "0"
log_callback(f"[Debug] 资料已填写,等待 Cloudflare 人机验证通过... 当前token长度={token_len}")
if token_len == "0":
pause_seconds = random.uniform(1, 3)
if log_callback:
log_callback(f"[Debug] Cloudflare token 为空,暂停 {pause_seconds:.1f}s 后继续检测")
sleep_with_cancel(pause_seconds, cancel_callback)
now = time.time() now = time.time()
if wait_cf_since is None: if wait_cf_since is None:
wait_cf_since = now wait_cf_since = now
token_len = filled.split(":", 1)[1] if ":" in filled else "0" # 卡住后自动二次复用 Turnstile 组件
if log_callback and (last_cf_log_at <= 0 or now - last_cf_log_at >= 5): if now - wait_cf_since >= 12 and now - last_cf_retry_at >= 8:
waited = now - wait_cf_since if wait_cf_since else 0 if log_callback:
log_callback( log_callback("[Debug] Cloudflare 验证卡住,开始二次复用 Turnstile...")
f"[*] 资料已填写,等待 Cloudflare 人机验证通过... "
f"当前token长度={token_len} waited={waited:.0f}s"
)
last_cf_log_at = now
# Git-creat7: token 为空时多等 1-3s
if token_len == "0":
pause_seconds = random.uniform(1.0, 3.0)
if log_callback and (now - last_cf_log_at >= 4):
log_callback(f"[*] Cloudflare token 为空,暂停 {pause_seconds:.1f}s 后继续检测")
last_cf_log_at = now
sleep_with_cancel(pause_seconds, cancel_callback)
# Cookie overlay can reappear mid-wait and keep iframe=0
if wait_cf_since and (now - wait_cf_since) >= 2.0:
try: try:
dismiss_cookie_banner(page, log_callback=None) token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback)
except Exception: if token:
pass synced = page.run_js(
# 轻点一次(参考会在 getTurnstileToken 内反复点;外层也补一次) """
if now - wait_cf_since >= 3.0 and now - last_cf_retry_at >= 3.0: const token = String(arguments[0] || '').trim();
try: const cfInput = document.querySelector('input[name="cf-turnstile-response"]');
_click_turnstile_widget(page) if (!cfInput || !token) return false;
except Exception: const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set;
pass if (nativeSetter) nativeSetter.call(cfInput, token);
# 卡住后二次复用 Turnstile 组件(reset + shadow click) else cfInput.value = token;
if now - wait_cf_since >= cf_retry_after and now - last_cf_retry_at >= cf_retry_gap: cfInput.dispatchEvent(new Event('input', { bubbles: true }));
_retry_turnstile("Cloudflare 验证卡住,开始二次复用 Turnstile...") cfInput.dispatchEvent(new Event('change', { bubbles: true }));
return String(cfInput.value || '').trim().length;
""",
token,
)
if log_callback:
log_callback(f"[Debug] Turnstile 二次复用完成,回填长度={synced}")
except Exception as cf_exc:
if log_callback:
log_callback(f"[Debug] Turnstile 二次复用失败: {cf_exc}")
last_cf_retry_at = now
sleep_with_cancel(0.8, cancel_callback) sleep_with_cancel(0.8, cancel_callback)
continue continue
@@ -4737,10 +4377,10 @@ return 'filled-no-submit';
form_filled_once = True form_filled_once = True
elif filled == "fill-failed" and log_callback: elif filled == "fill-failed" and log_callback:
log_callback("[Debug] 资料输入失败,重试中...") log_callback("[Debug] 资料输入失败,重试中...")
human_sleep(0.5, cancel_callback) sleep_with_cancel(0.5, cancel_callback)
continue continue
elif filled == "not-ready": elif filled == "not-ready":
human_sleep(0.5, cancel_callback) sleep_with_cancel(0.5, cancel_callback)
continue continue
submit_state = page.run_js( submit_state = page.run_js(
@@ -4776,7 +4416,7 @@ const buttons = Array.from(document.querySelectorAll('button[type="submit"], but
}); });
const submitBtn = buttons.find((node) => { const submitBtn = buttons.find((node) => {
const t = buttonText(node).replace(/\s+/g, '').toLowerCase(); const t = buttonText(node).replace(/\s+/g, '').toLowerCase();
return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount') || t.includes('completesignup'); return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount');
}); });
if (!submitBtn) { if (!submitBtn) {
const visibleTexts = buttons.map(buttonText).filter(Boolean).slice(0, 8).join(' | '); const visibleTexts = buttons.map(buttonText).filter(Boolean).slice(0, 8).join(' | ');
@@ -4789,43 +4429,55 @@ return 'submitted';
) )
if isinstance(submit_state, str) and submit_state.startswith("wait-cloudflare"): if isinstance(submit_state, str) and submit_state.startswith("wait-cloudflare"):
if log_callback:
token_len = submit_state.split(":", 1)[1] if ":" in submit_state else "0"
log_callback(f"[Debug] 等待 Cloudflare 人机验证通过后再提交... 当前token长度={token_len}")
now = time.time() now = time.time()
if wait_cf_since is None: if wait_cf_since is None:
wait_cf_since = now wait_cf_since = now
if log_callback and (last_cf_log_at <= 0 or now - last_cf_log_at >= 5): if now - wait_cf_since >= 12 and now - last_cf_retry_at >= 8:
token_len = submit_state.split(":", 1)[1] if ":" in submit_state else "0" if log_callback:
waited = now - wait_cf_since if wait_cf_since else 0 log_callback("[Debug] 提交前仍卡住,自动再次复用 Turnstile...")
log_callback(
f"[*] 等待 Cloudflare 人机验证通过后再提交... "
f"当前token长度={token_len} waited={waited:.0f}s"
)
last_cf_log_at = now
if wait_cf_since and now - wait_cf_since >= 3.0 and now - last_cf_retry_at >= 3.0:
try: try:
_click_turnstile_widget(page) token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback)
except Exception: if token:
pass synced = page.run_js(
if now - wait_cf_since >= cf_retry_after and now - last_cf_retry_at >= cf_retry_gap: """
_retry_turnstile("提交前仍卡住,自动再次复用 Turnstile...") const token = String(arguments[0] || '').trim();
const cfInput = document.querySelector('input[name="cf-turnstile-response"]');
if (!cfInput || !token) return false;
const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set;
if (nativeSetter) nativeSetter.call(cfInput, token);
else cfInput.value = token;
cfInput.dispatchEvent(new Event('input', { bubbles: true }));
cfInput.dispatchEvent(new Event('change', { bubbles: true }));
return String(cfInput.value || '').trim().length;
""",
token,
)
if log_callback:
log_callback(f"[Debug] Turnstile 二次复用完成,回填长度={synced}")
except Exception as cf_exc:
if log_callback:
log_callback(f"[Debug] Turnstile 二次复用失败: {cf_exc}")
last_cf_retry_at = now
sleep_with_cancel(0.8, cancel_callback) sleep_with_cancel(0.8, cancel_callback)
continue continue
if submit_state == "submitted": if submit_state == "submitted":
if log_callback: if log_callback:
log_callback(f"[*] 已填写注册资料并提交: {given_name} {family_name}") log_callback(f"[Debug] 已填写注册资料并提交: {given_name} {family_name}")
return {"given_name": given_name, "family_name": family_name, "password": password} return {"given_name": given_name, "family_name": family_name, "password": password}
wait_cf_since = None wait_cf_since = None
if isinstance(submit_state, str) and submit_state.startswith("no-submit-button") and log_callback: if isinstance(submit_state, str) and submit_state.startswith("no-submit-button") and log_callback:
log_callback(f"[Debug] 未找到提交按钮,继续等待页面稳定... {submit_state[17:80]}") visible_buttons = submit_state.split(":", 1)[1] if ":" in submit_state else ""
suffix = f" 可见按钮: {visible_buttons}" if visible_buttons else ""
log_callback(f"[Debug] 未找到提交按钮,继续等待页面稳定...{suffix}")
human_sleep(0.5, cancel_callback) sleep_with_cancel(0.5, cancel_callback)
raise Exception("最终注册页资料填写失败") raise Exception("最终注册页资料填写失败")
# ── NSFW 自动开启(从 AaronL725 移植) ──
def generate_random_birthdate(): def generate_random_birthdate():
"""生成随机生日(20-40 岁)。""" """生成随机生日(20-40 岁)。"""
import datetime as dt import datetime as dt
+345
View File
@@ -0,0 +1,345 @@
#!/usr/bin/env python3
"""Replace local Turnstile/browser CF path with pure zip implementations.
Keeps local Gmail / hotmail / register_cli / cpa_xai intact.
Source of truth: grok-register.zip (ReinerBRO) for CF + browser options.
"""
from __future__ import annotations
import os
import re
import shutil
from pathlib import Path
ROOT = Path(r"D:/Projects/AI/grok-keygen")
# Git Bash /tmp maps to %TEMP% on Windows
_TEMP = Path(os.environ.get("TEMP") or os.environ.get("TMP") or r"C:/Users/Chaos/AppData/Local/Temp")
ZIP_REG = _TEMP / "zip_full/grok-register/grok_register_ttk.py"
ZIP_CPA = _TEMP / "zip_full/grok_reg-protocol_cpa/grok_register_ttk.py"
ZIP_PATCH = _TEMP / "zip_full/grok-register/turnstilePatch"
LOCAL = ROOT / "grok_register_ttk.py"
LOCAL_PATCH = ROOT / "turnstilePatch"
def extract_func(text: str, name: str) -> str:
lines = text.splitlines(keepends=True)
starts = {}
for i, line in enumerate(lines):
m = re.match(r"^def (\w+)\(", line)
if m:
starts[m.group(1)] = i
m = re.match(r"^([A-Z_][A-Z0-9_]*)\s*=", line)
if m:
starts[m.group(1)] = i
if name not in starts:
raise KeyError(f"missing {name}")
s = starts[name]
e = len(lines)
for j in range(s + 1, len(lines)):
if re.match(r"^(def |class |[A-Z_][A-Z0-9_]*\s*=)", lines[j]):
e = j
break
return "".join(lines[s:e]).rstrip() + "\n\n\n"
def replace_func(src: str, name: str, new_body: str) -> str:
"""Replace a top-level def/assignment block by name."""
lines = src.splitlines(keepends=True)
start = None
for i, line in enumerate(lines):
if re.match(rf"^def {re.escape(name)}\(", line) or re.match(
rf"^{re.escape(name)}\s*=", line
):
start = i
break
if start is None:
raise KeyError(f"local missing {name}")
end = len(lines)
for j in range(start + 1, len(lines)):
if re.match(r"^(def |class |[A-Z_][A-Z0-9_]*\s*=)", lines[j]):
end = j
break
# preserve trailing blank style: ensure exactly two newlines before next
body = new_body
if not body.endswith("\n"):
body += "\n"
# normalize to end with two blank lines if next block exists
while body.endswith("\n\n\n"):
body = body[:-1]
if not body.endswith("\n\n"):
body = body.rstrip("\n") + "\n\n"
return "".join(lines[:start]) + body + "".join(lines[end:])
def remove_func(src: str, name: str) -> str:
lines = src.splitlines(keepends=True)
start = None
for i, line in enumerate(lines):
if re.match(rf"^def {re.escape(name)}\(", line) or re.match(
rf"^{re.escape(name)}\s*=", line
):
start = i
break
if start is None:
return src
end = len(lines)
for j in range(start + 1, len(lines)):
if re.match(r"^(def |class |[A-Z_][A-Z0-9_]*\s*=)", lines[j]):
end = j
break
return "".join(lines[:start]) + "".join(lines[end:])
def adapt_reg_get_turnstile(zip_fn: str) -> str:
"""grok-register uses session.page; local uses _get_page()."""
text = zip_fn
text = text.replace(
"def getTurnstileToken(session, log_callback=None, cancel_callback=None):\n page = session.page\n",
"def getTurnstileToken(log_callback=None, cancel_callback=None, max_rounds: int = 20):\n page = _get_page()\n",
)
# support max_rounds
text = text.replace(
"for _ in range(0, 20):",
"rounds = max(1, int(max_rounds or 20))\n for _ in range(0, rounds):",
)
# local uses human_sleep/cf_sleep; zip uses sleep_with_cancel — keep sleep_with_cancel (exists)
return text
def adapt_reg_fill_profile(zip_fn: str) -> str:
"""Adapt fill_profile from grok-register to local globals."""
text = zip_fn
text = text.replace(
"def fill_profile_and_submit(session, timeout=120, log_callback=None, cancel_callback=None):\n page = session.page\n",
"def fill_profile_and_submit(timeout=120, log_callback=None, cancel_callback=None):\n page = _get_page()\n check_timeout(time.time())\n dump_state(page, \"profile-form\")\n take_screenshot(page, \"profile-form\")\n",
)
# fix getTurnstileToken(session, ...) -> getTurnstileToken(...)
text = text.replace(
"token = getTurnstileToken(session, log_callback=log_callback, cancel_callback=cancel_callback)",
"token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback)",
)
# add zip warmup like protocol_cpa
if "预热 Turnstile" not in text:
text = text.replace(
"given_name, family_name, password = build_profile()\n deadline = time.time() + timeout\n",
"given_name, family_name, password = build_profile()\n"
" if log_callback:\n"
" log_callback(\"[*] 预热 Turnstile...\")\n"
" human_sleep(2, cancel_callback)\n"
" deadline = time.time() + timeout\n",
)
return text
def adapt_create_browser_options(zip_fn: str) -> str:
"""Use zip minimal options; keep local headless/proxy helpers lightly."""
# Prefer ReinerBRO pure options (no disable-images). Keep proxy + headless glue.
return '''def create_browser_options(headless=None):
"""Zip-style Chromium options (grok-register / ReinerBRO).
Critical for CF:
- always load turnstilePatch
- no --disable-images
- no browser_path unless config/env set
- no AutomationControlled / heavy stealth flags on headed
"""
options = ChromiumOptions()
options.auto_port()
options.set_timeouts(base=1)
# ReinerBRO headed flags
for flag in (
"--no-first-run",
"--no-default-browser-check",
"--disable-background-networking",
"--disable-component-update",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
):
options.set_argument(flag)
use_headless = should_use_headless(headless, log_hint=True)
if use_headless:
apply_headless_to_options(options, True)
for flag in (
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
"--mute-audio",
):
options.set_argument(flag)
print(" [browser] headless=true", flush=True)
else:
print(" [browser] headless=false (zip headed)", flush=True)
# system default unless override
browser_path = resolve_browser_path()
if browser_path:
try:
options.set_browser_path(browser_path)
print(f" [browser] path={browser_path}", flush=True)
except Exception as exc:
print(f" [browser] set path failed: {exc}", flush=True)
else:
print(" [browser] path=system-default (zip style)", flush=True)
# Force UA from config (Chrome/138 etc.)
try:
ua = get_user_agent()
if ua:
options.set_user_agent(ua)
print(f" [browser] ua={ua[:60]}...", flush=True)
except Exception:
pass
if os.path.exists(EXTENSION_PATH):
try:
options.add_extension(EXTENSION_PATH)
print(f" [ext] turnstilePatch loaded: {EXTENSION_PATH}", flush=True)
except Exception as ext_exc:
print(f" [ext] turnstilePatch load failed: {ext_exc}", flush=True)
else:
print(f" [ext] turnstilePatch missing: {EXTENSION_PATH}", flush=True)
proxy = (config.get("proxy") or "").strip() if isinstance(config, dict) else ""
if proxy:
try:
from urllib.parse import urlparse
u = urlparse(proxy if "://" in proxy else f"http://{proxy}")
host = u.hostname or ""
if host:
port = u.port or (443 if (u.scheme or "http") == "https" else 80)
scheme = u.scheme or "http"
options.set_argument(f"--proxy-server={scheme}://{host}:{port}")
print(f" [proxy] browser --proxy-server={scheme}://{host}:{port}", flush=True)
except Exception as e:
print(f" [proxy] set browser proxy failed: {e}", flush=True)
return options
'''
def main() -> None:
if not ZIP_REG.is_file():
raise SystemExit(f"missing {ZIP_REG}")
reg_text = ZIP_REG.read_text(encoding="utf-8", errors="replace")
local = LOCAL.read_text(encoding="utf-8", errors="replace")
# backup
bak = ROOT / "grok_register_ttk.py.bak_pre_zip_port"
shutil.copy2(LOCAL, bak)
print(f"backup -> {bak}")
# restore pure zip turnstilePatch
if ZIP_PATCH.is_dir():
LOCAL_PATCH.mkdir(exist_ok=True)
for name in ("content.js", "manifest.json"):
src = ZIP_PATCH / name
if src.is_file():
shutil.copy2(src, LOCAL_PATCH / name)
print(f"copied {name}")
# remove MV3 leftovers
for stale in ("script.js",):
p = LOCAL_PATCH / stale
if p.exists():
p.unlink()
print(f"removed {stale}")
get_ts = adapt_reg_get_turnstile(extract_func(reg_text, "getTurnstileToken"))
fill = adapt_reg_fill_profile(extract_func(reg_text, "fill_profile_and_submit"))
create_opts = adapt_create_browser_options("")
# Remove interactive-safe helpers that zip doesn't need (keep if referenced elsewhere carefully)
for helper in (
"_click_turnstile_widget",
"_read_turnstile_token_simple",
"_turnstile_ui_state",
"_read_turnstile_token",
"_turnstile_diag",
):
before = local
local = remove_func(local, helper)
if local != before:
print(f"removed helper {helper}")
local = replace_func(local, "getTurnstileToken", get_ts)
print("replaced getTurnstileToken")
local = replace_func(local, "fill_profile_and_submit", fill)
print("replaced fill_profile_and_submit")
local = replace_func(local, "create_browser_options", create_opts)
print("replaced create_browser_options")
# Simplify resolve_browser_path to zip style (auto -> None)
resolve_browser = '''def resolve_browser_path():
"""Zip style: only return path if config/env override exists.
browser_prefer=auto/system → None (DrissionPage system Chrome).
"""
override = str((config.get("browser_path") if isinstance(config, dict) else "") or "").strip()
if override:
norm = _normalize_browser_path(override)
if norm:
return norm
print(f" [browser] ignore invalid browser_path={override!r}", flush=True)
for env_key in ("BROWSER_PATH", "CHROME_PATH", "CHROMIUM_PATH"):
env_path = (os.environ.get(env_key) or "").strip()
if not env_path:
continue
norm = _normalize_browser_path(env_path)
if norm:
return norm
print(f" [browser] ignore invalid {env_key}={env_path!r}", flush=True)
prefer = str((config.get("browser_prefer") if isinstance(config, dict) else "") or "auto").strip().lower()
if prefer in ("", "default", "detect", "system"):
prefer = "auto"
env_pref = (os.environ.get("BROWSER_PREFER") or "").strip().lower()
if env_pref:
prefer = env_pref
if prefer in ("auto", "system"):
return None
if prefer in ("chrome", "google-chrome", "google"):
order = list(CHROME_CANDIDATES) + list(CHROMIUM_CANDIDATES)
elif prefer in ("chromium", "chromeium"):
order = list(CHROMIUM_CANDIDATES) + list(CHROME_CANDIDATES)
else:
return None
return _first_existing_browser(order)
'''
local = replace_func(local, "resolve_browser_path", resolve_browser)
print("replaced resolve_browser_path")
# Force stealth off by default in apply_stealth_patches
local = re.sub(
r'mode = str\(\(config\.get\("stealth_mode"\).*?or "\w+"\)\.strip\(\)\.lower\(\)',
'mode = str((config.get("stealth_mode") if isinstance(config, dict) else "") or "off").strip().lower()',
local,
count=1,
)
# Ensure fill_profile / other call sites don't reference removed helpers
# getTurnstileToken(..., max_rounds=) still ok
if "_turnstile_ui_state" in local or "_click_turnstile_widget" in local:
print("WARN: leftover helper references:")
for i, line in enumerate(local.splitlines(), 1):
if "_turnstile_ui_state" in line or "_click_turnstile_widget" in line or "_read_turnstile_token" in line:
print(f" L{i}: {line.strip()[:120]}")
LOCAL.write_text(local, encoding="utf-8")
print(f"wrote {LOCAL} ({len(local.splitlines())} lines)")
# syntax check
import py_compile
py_compile.compile(str(LOCAL), doraise=True)
print("syntax OK")
if __name__ == "__main__":
main()
+114
View File
@@ -0,0 +1,114 @@
// Turnstile Patch - 隐藏自动化标识,加速 Turnstile 验证
// 在 document_start 阶段执行,确保在页面脚本之前生效
(function () {
"use strict";
// 1. 隐藏 navigator.webdriver 标识
// Chrome 自动化模式下 navigator.webdriver = true,Turnstile 会检测此属性
try {
Object.defineProperty(navigator, "webdriver", {
get: function () {
return false;
},
configurable: true,
});
} catch (e) {}
// 2. 移除 Chrome 自动化相关的 Runtime 属性
try {
if (window.chrome && window.chrome.runtime) {
delete window.chrome.runtime.onConnect;
delete window.chrome.runtime.onMessage;
}
} catch (e) {}
// 3. 覆盖 permissions.query,隐藏 notifications 权限异常
try {
var origQuery = navigator.permissions.query.bind(navigator.permissions);
navigator.permissions.query = function (params) {
if (params.name === "notifications") {
return Promise.resolve({ state: Notification.permission });
}
return origQuery(params);
};
} catch (e) {}
// 4. 修补 plugin 数量,模拟正常浏览器
try {
Object.defineProperty(navigator, "plugins", {
get: function () {
return [1, 2, 3, 4, 5];
},
configurable: true,
});
} catch (e) {}
// 5. 修补 languages 属性
try {
Object.defineProperty(navigator, "languages", {
get: function () {
return ["en-US", "en"];
},
configurable: true,
});
} catch (e) {}
// 6. 页面加载完成后,自动监控并点击 Turnstile 复选框
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", autoClickTurnstile);
} else {
autoClickTurnstile();
}
function autoClickTurnstile() {
// 定时检查 Turnstile iframe 是否出现
var checkCount = 0;
var maxChecks = 100; // 最多检查 100 次(约 50 秒)
var timer = setInterval(function () {
checkCount++;
if (checkCount > maxChecks) {
clearInterval(timer);
return;
}
try {
// 查找 Turnstile iframe
var iframes = document.querySelectorAll(
'iframe[src*="challenges.cloudflare.com"], iframe[src*="turnstile"]'
);
for (var i = 0; i < iframes.length; i++) {
var iframe = iframes[i];
try {
// 尝试访问 iframe 内部的 checkbox
var body = iframe.contentDocument || iframe.contentWindow.document;
var checkbox = body.querySelector(
'input[type="checkbox"], .mark, #cf-chl-widget-nomu1_resp'
);
if (checkbox && !checkbox.checked) {
checkbox.click();
}
} catch (e) {
// 跨域限制,尝试通过 postMessage 触发
try {
iframe.contentWindow.postMessage(
{ type: "turnstile-auto-click" },
"*"
);
} catch (e2) {}
}
}
// 也尝试直接操作 Turnstile API
if (
window.turnstile &&
typeof window.turnstile.getResponse === "function"
) {
var resp = window.turnstile.getResponse();
if (resp && resp.length > 0) {
clearInterval(timer); // 已获得 token,停止检查
}
}
} catch (e) {}
}, 500);
}
})();
+8 -9
View File
@@ -1,16 +1,15 @@
{ {
"manifest_version": 3, "manifest_version": 2,
"name": "Turnstile Patcher", "name": "Turnstile Patch",
"version": "2.1", "version": "1.0.0",
"description": "Patch CDP MouseEvent.screenX/screenY for Turnstile (ReinerBRO / Git-creat7)", "description": "Patch browser automation detection for Turnstile",
"content_scripts": [ "content_scripts": [
{ {
"js": ["./script.js"],
"matches": ["<all_urls>"], "matches": ["<all_urls>"],
"js": ["content.js"],
"run_at": "document_start", "run_at": "document_start",
"all_frames": true, "all_frames": true
"world": "MAIN",
"match_about_blank": true
} }
] ],
"permissions": ["activeTab"]
} }
-24
View File
@@ -1,24 +0,0 @@
function getRandomInt(min, max) {
return Math.floor(Math.random() * (max - min + 1)) + min;
}
// CDP Input.dispatchMouseEvent creates MouseEvent where
// .screenX/.screenY equal .clientX/.clientY. Cloudflare Turnstile detects
// this (Chromium issue 40280325). Patch with realistic screen coords.
// Source: ObjectAscended/CDP-bug-MouseEvent-.screenX-.screenY-patcher
// Ported from ReinerBRO/grok-register + Git-creat7/grokRegister-cpa.
// old method wouldn't work on 4k screens
let screenX = getRandomInt(800, 1200);
let screenY = getRandomInt(400, 600);
Object.defineProperty(MouseEvent.prototype, "screenX", { value: screenX });
Object.defineProperty(MouseEvent.prototype, "screenY", { value: screenY });
// Best-effort: also cover PointerEvent if present (CF may read either).
try {
if (typeof PointerEvent !== "undefined" && PointerEvent.prototype) {
Object.defineProperty(PointerEvent.prototype, "screenX", { value: screenX });
Object.defineProperty(PointerEvent.prototype, "screenY", { value: screenY });
}
} catch (e) {}