From 4abdc8aa4a08516d331ca8a4df04e218a7d62ed9 Mon Sep 17 00:00:00 2001 From: Chaos Date: Tue, 14 Jul 2026 14:06:13 +0800 Subject: [PATCH] Revert mint path to pre-browser PKCE baseline (2e833f2). Restore HTTP-only SSO OAuth with curl_cffi then std requests fallback; remove browser PKCE prefer/fallback knobs and mint_from_sso_browser. --- config.json | 3 - cpa_export.py | 68 +--- grok_register_ttk.py | 29 +- oidc_mint/__init__.py | 2 - oidc_mint/oauth_code.py | 700 ++-------------------------------------- 5 files changed, 51 insertions(+), 751 deletions(-) diff --git a/config.json b/config.json index 59adbea..fe5fd8a 100644 --- a/config.json +++ b/config.json @@ -40,9 +40,6 @@ "cpa_push_required": false, "cpa_prefer_sso_oauth": true, "cpa_require_referrer": true, - "cpa_prefer_browser_mint": false, - "cpa_allow_browser_fallback": false, - "cpa_browser_mint_timeout_sec": 90, "cpa_allow_device_fallback": false, "mint_proxy": "", "mint_timeout_sec": 300, diff --git a/cpa_export.py b/cpa_export.py index e93f873..9a359d4 100644 --- a/cpa_export.py +++ b/cpa_export.py @@ -72,49 +72,13 @@ def _mint_tokens( log: Callable[[str], None], proxy: str | None, ) -> dict[str, Any]: - """铸造策略（仅 HTTP；浏览器/设备码默认关）： - - 1. 仅当 cpa_prefer_browser_mint=true 且有 page：先浏览器 PKCE - 2. HTTP SSO→OAuth（curl_cffi，短超时）— 默认唯一主路径 - 3. 仅当 cpa_allow_browser_fallback=true 且 HTTP 失败：浏览器 PKCE - 4. 可选设备码（通常无 referrer，默认关） - """ - from oidc_mint.oauth_code import ( - OAuthCodeError, - _is_http_tls_failure, - mint_from_sso, - mint_from_sso_browser, - normalize_sso_cookie, - ) + """优先 SSO 授权码；可选回退设备码。""" + from oidc_mint.oauth_code import OAuthCodeError, mint_from_sso, normalize_sso_cookie sso_token = normalize_sso_cookie(sso or "") prefer_sso = bool(cfg.get("cpa_prefer_sso_oauth", True)) - allow_browser = bool(cfg.get("cpa_allow_browser_fallback", False)) - # 默认 False：不走浏览器优先 - prefer_browser = bool(cfg.get("cpa_prefer_browser_mint", False)) allow_device = bool(cfg.get("cpa_allow_device_fallback", False)) timeout = float(cfg.get("mint_timeout_sec", 300) or 300) - require_ref = bool(cfg.get("cpa_require_referrer", True)) - http_err: Exception | None = None - browser_timeout = min(timeout, float(cfg.get("cpa_browser_mint_timeout_sec", 90) or 90)) - - def _browser_mint(reason: str) -> dict[str, Any]: - log(f"[cpa] 浏览器 PKCE 铸造（{reason}）") - return mint_from_sso_browser( - sso_token, - page, - log=lambda m: log(f"[Debug] {m}"), - require_referrer=require_ref, - timeout_sec=browser_timeout, - ) - - # 路径 A：有 page 时优先浏览器（最稳，不依赖 Python→auth.x.ai 直连） - if prefer_sso and sso_token and allow_browser and page is not None and prefer_browser: - try: - return _browser_mint("优先浏览器，绕开 Python 直连 auth.x.ai") - except Exception as exc: # noqa: BLE001 - log(f"[!] 浏览器 PKCE 优先路径失败: {exc}") - log("[cpa] 继续尝试 HTTP SSO→OAuth") if prefer_sso and sso_token: log("[cpa] 使用 SSO→OAuth(PKCE, referrer=grok-build)") @@ -123,35 +87,21 @@ def _mint_tokens( sso_token, proxy=proxy, log=lambda m: log(f"[Debug] {m}"), - require_referrer=require_ref, + require_referrer=bool(cfg.get("cpa_require_referrer", True)), ) except OAuthCodeError as exc: - http_err = exc log(f"[!] SSO→OAuth 失败: {exc}") + if not allow_device: + raise + log("[cpa] 回退设备码铸造（可能缺 referrer）") except Exception as exc: # noqa: BLE001 - http_err = exc log(f"[!] SSO→OAuth 异常: {exc}") - - # HTTP 失败后：有 page+sso 再试浏览器（若优先路径没走过或当时失败） - if allow_browser and page is not None and sso_token and http_err is not None: - why = "TLS/连接/超时" if _is_http_tls_failure(http_err) else "HTTP" - try: - return _browser_mint(f"{why}失败后回退") - except Exception as exc: # noqa: BLE001 - log(f"[!] 浏览器 PKCE 失败: {exc}") - if not allow_device: - raise - log("[cpa] 继续回退设备码铸造（可能缺 referrer）") - elif http_err is not None and not allow_device: - raise http_err + if not allow_device: + raise + log("[cpa] 回退设备码铸造（可能缺 referrer）") if not allow_device and not sso_token: raise RuntimeError("无 sso cookie，且已禁用设备码回退；无法铸造带 referrer 的 token") - if not allow_device: - # 有 sso 但 browser 也没开/没 page - if http_err is not None: - raise http_err - raise RuntimeError("SSO 铸造失败，且未启用任何回退") # 设备码回退（旧路径，通常无 referrer） from oidc_mint import mint_with_browser diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 431ac96..ba3e984 100644 --- a/grok_register_ttk.py +++ b/grok_register_ttk.py @@ -85,9 +85,6 @@ DEFAULT_CONFIG = { # OIDC：优先 SSO→Authorization Code + referrer=grok-build "cpa_prefer_sso_oauth": True, # True=用 sso cookie 走 PKCE（必须带 referrer） "cpa_require_referrer": True, # True=access_token 无 referrer=grok-build 则失败 - "cpa_prefer_browser_mint": False, # True=有 page 时先浏览器（慢）；默认关 - "cpa_allow_browser_fallback": False, # True=HTTP 失败再用浏览器 PKCE；默认关（不要网页铸造） - "cpa_browser_mint_timeout_sec": 90, # 浏览器 PKCE 最长等待（仅 fallback 开启时） "cpa_allow_device_fallback": False, # True=SSO 失败时回退设备码（通常不可用） # OIDC 铸造代理/超时 "mint_proxy": "", # 铸造专用代理；空=复用 proxy @@ -2098,7 +2095,6 @@ def update_nsfw_settings(session, log_callback=None): def _is_curl_tls_error(exc) -> bool: - """curl_cffi 库损坏 / TLS / 连接超时：可尝试换后端或记失败，勿当业务错误。""" text = str(exc or "").lower() return any( n in text @@ -2106,16 +2102,9 @@ def _is_curl_tls_error(exc) -> bool: "openssl_internal:invalid library", "tls connect error", "curl: (35)", - "curl: (28)", "failed to perform, curl: (35)", - "failed to perform, curl: (28)", - "connection timed out", "ssl_error_syscall", "ssl connect error", - "readtimeout", - "connecttimeout", - "timed out", - "timeout", ) ) @@ -2176,18 +2165,24 @@ def enable_nsfw_for_token(token, cf_clearance="", log_callback=None): except Exception: pass - # 仅用 curl_cffi：本机实测 std requests 对 accounts.x.ai/auth.x.ai 更易 ReadTimeout try: session = _make_post_reg_session(proxies, prefer="curl") try: return _run(session) except Exception as exc: - if _is_curl_tls_error(exc): - log(f"[Debug] set_tos/nsfw 网络/TLS 失败（不回退 requests）: {exc}") - return False, f"网络/TLS: {exc}" - return False, f"异常: {exc}" - finally: + if not _is_curl_tls_error(exc): + return False, f"异常: {exc}" + log(f"[Debug] curl TLS 异常，set_tos/nsfw 回退 requests: {exc}") _close(session) + session = _make_post_reg_session(proxies, prefer="requests") + try: + return _run(session) + finally: + _close(session) + session = None + finally: + if session is not None: + _close(session) except Exception as e: return False, f"异常: {str(e)}" diff --git a/oidc_mint/__init__.py b/oidc_mint/__init__.py index 60e48b0..ec7c5c2 100644 --- a/oidc_mint/__init__.py +++ b/oidc_mint/__init__.py @@ -16,7 +16,6 @@ from .oauth_code import ( OAuthCodeError, SCOPE as CODE_SCOPE, mint_from_sso, - mint_from_sso_browser, normalize_sso_cookie, sso_to_token, ) @@ -27,7 +26,6 @@ __all__ = [ "mint_with_browser", "shutdown_mint_browsers", "mint_from_sso", - "mint_from_sso_browser", "sso_to_token", "normalize_sso_cookie", "CLIENT_ID", diff --git a/oidc_mint/oauth_code.py b/oidc_mint/oauth_code.py index 76186d2..2199eb5 100644 --- a/oidc_mint/oauth_code.py +++ b/oidc_mint/oauth_code.py @@ -260,10 +260,6 @@ def _final_url(resp: Any) -> str: return "" -# HTTP 铸造单步超时：直连/跨境链路差时不要卡 30s，尽快让上层走浏览器 -HTTP_STEP_TIMEOUT = 12 - - def open_authorize_page(session: Any, flow: AuthCodeFlow) -> str: params = { "response_type": "code", @@ -281,7 +277,7 @@ def open_authorize_page(session: Any, flow: AuthCodeFlow) -> str: url, headers=_browser_headers("GET", url), allow_redirects=True, - timeout=HTTP_STEP_TIMEOUT, + timeout=30, ) body = resp.text or "" final = _final_url(resp) @@ -359,7 +355,7 @@ def approve_authorization(session: Any, consent_url: str, flow: AuthCodeFlow) -> data=body.encode("utf-8"), headers=_browser_headers("POST", consent_url, NEXT_ACTION_ID), allow_redirects=True, - timeout=HTTP_STEP_TIMEOUT, + timeout=30, ) text = resp.text or "" if resp.status_code < 200 or resp.status_code >= 300: @@ -388,7 +384,7 @@ def exchange_auth_code(session: Any, code: str, flow: AuthCodeFlow) -> TokenResu TOKEN_URL, data=urlencode(form), headers=_token_headers(), - timeout=HTTP_STEP_TIMEOUT, + timeout=30, ) text = resp.text or "" if resp.status_code < 200 or resp.status_code >= 300: @@ -458,27 +454,47 @@ def sso_to_token( log: LogFn | None = None, require_referrer: bool = True, ) -> TokenResult: - """SSO cookie → 带 referrer=grok-build 的 OAuth token。 - - 仅用 curl_cffi（Chrome TLS）。不再回退 std requests： - 实测 requests 访问 auth.x.ai / accounts.x.ai 常 ReadTimeout，比 curl 更差。 - 连接/TLS/超时由上层切到浏览器 PKCE。 - """ + """SSO cookie → 带 referrer=grok-build 的 OAuth token。""" log = log or _noop_log sso = normalize_sso_cookie(sso_cookie) if not sso: raise OAuthCodeError("sso cookie 为空") + # 先 curl_cffi；若遇到 OpenSSL invalid library / curl(35)，自动回退 std requests session = _make_session(proxy, prefer="curl") try: return _run_sso_flow( sso, session=session, log=log, require_referrer=require_referrer ) - finally: + except Exception as exc: # noqa: BLE001 + backend = str(getattr(session, "_cpa_http_backend", "") or "") + can_fallback = _is_curl_tls_broken(exc) or ( + backend.startswith("curl_cffi") and "curl: (35)" in str(exc).lower() + ) + if not can_fallback: + raise + log(f"curl TLS 异常，回退标准 requests: {_short(str(exc), 160)}") try: session.close() except Exception: pass + session = _make_session(proxy, prefer="requests") + try: + return _run_sso_flow( + sso, session=session, log=log, require_referrer=require_referrer + ) + finally: + try: + session.close() + except Exception: + pass + session = None # type: ignore[assignment] + finally: + if session is not None: + try: + session.close() + except Exception: + pass def mint_from_sso( @@ -504,659 +520,3 @@ def mint_from_sso( "referrer": tr.referrer, "sso": normalize_sso_cookie(sso_cookie), } - - -def _is_http_tls_failure(exc: BaseException | str) -> bool: - """HTTP 层 TLS/连接/超时失败：适合改走浏览器铸造。""" - text = str(exc or "").lower() - needles = ( - "unexpected_eof_while_reading", - "sslerror", - "ssleoferror", - "max retries exceeded", - "openssl_internal:invalid library", - "tls connect error", - "curl: (35)", - "curl: (28)", - "failed to perform, curl: (35)", - "failed to perform, curl: (28)", - "connection timed out", - "ssl_error_syscall", - "connection reset", - "connection aborted", - "name resolution", - "readtimeout", - "connecttimeout", - "timed out", - "timeout", - ) - return any(n in text for n in needles) - - -def _page_eval(page: Any, js: str, *args: Any) -> Any: - """兼容 DrissionPage page.run_js / page.run_js_loaded。""" - if page is None: - raise OAuthCodeError("page 为空，无法浏览器铸造") - last_err: Exception | None = None - for name in ("run_js", "run_js_loaded", "run_async_js"): - fn = getattr(page, name, None) - if not callable(fn): - continue - try: - if args: - return fn(js, *args) - return fn(js) - except TypeError: - # 某些签名不接受额外参数 - try: - return fn(js) - except Exception as exc: # noqa: BLE001 - last_err = exc - except Exception as exc: # noqa: BLE001 - last_err = exc - continue - raise OAuthCodeError(f"page 无法执行 JS: {last_err or 'no run_js'}") - - -def _ensure_sso_on_page(page: Any, sso: str, log: LogFn) -> None: - sso = normalize_sso_cookie(sso) - if not sso: - raise OAuthCodeError("sso cookie 为空") - # 先落到 accounts 域，再写 cookie，避免 set 失败 - for url in ("https://accounts.x.ai/", "https://auth.x.ai/"): - try: - page.get(url) - time.sleep(0.25) - except Exception as exc: # noqa: BLE001 - log(f"open {url} warn: {exc}") - - items = [] - for domain in (".x.ai", "accounts.x.ai", ".accounts.x.ai", "auth.x.ai", ".auth.x.ai"): - for name in ("sso", "sso-rw"): - items.append( - { - "name": name, - "value": sso, - "domain": domain, - "path": "/", - "secure": True, - "sameSite": "None", - } - ) - - set_ok = False - # 优先 CDP/DrissionPage set.cookies（可写 httpOnly 域 cookie） - for target in (page, getattr(page, "browser", None)): - if target is None: - continue - try: - setter = getattr(target, "set", None) - cookies_fn = getattr(setter, "cookies", None) if setter is not None else None - if not callable(cookies_fn): - continue - try: - cookies_fn(items) - set_ok = True - log(f"browser sso cookie set bulk via {type(target).__name__}") - break - except Exception: - n = 0 - for it in items: - try: - cookies_fn(it) - n += 1 - except Exception: - pass - if n: - set_ok = True - log(f"browser sso cookie set one-by-one={n}") - break - except Exception: - continue - - # document.cookie 兜底（非 httpOnly） - set_js = r""" -(sso) => { - try { - const maxAge = 60 * 60 * 24 * 30; - const base = `; path=/; max-age=${maxAge}; SameSite=None; Secure`; - document.cookie = `sso=${sso}${base}; domain=.x.ai`; - document.cookie = `sso-rw=${sso}${base}; domain=.x.ai`; - document.cookie = `sso=${sso}${base}`; - document.cookie = `sso-rw=${sso}${base}`; - return document.cookie.includes('sso='); - } catch (e) { - return String(e); - } -} -""" - try: - ok = _page_eval(page, set_js, sso) - log(f"browser sso document.cookie: {ok!r}") - set_ok = set_ok or (ok is True) or (ok == True) or (str(ok).lower() == "true") - except Exception as exc: # noqa: BLE001 - log(f"browser sso document.cookie fail: {exc}") - - if not set_ok: - raise OAuthCodeError("写入 sso cookie 失败") - - -def _page_html(page: Any) -> str: - try: - html = str(getattr(page, "html", "") or "") - if html: - return html - except Exception: - pass - try: - return str(_page_eval(page, "() => document.documentElement.outerHTML") or "") - except Exception: - return "" - - -def _extract_next_action_id(html: str) -> str: - """从 consent 页 HTML 抽 Next-Action / Server Action id。""" - text = html or "" - patterns = ( - r'"next-action"\s*:\s*"([a-f0-9]{20,})"', - r'"actionId"\s*:\s*"([a-f0-9]{20,})"', - r'\$ACTION_ID_([a-f0-9]{20,})', - r'next-action["\']?\s*[:=]\s*["\']([a-f0-9]{20,})', - ) - for pat in patterns: - m = re.search(pat, text, re.I) - if m: - return m.group(1) - return NEXT_ACTION_ID - - -def _browser_click_allow(page: Any, log: LogFn) -> bool: - """Consent「允许」必须真实点击（JS click 会导致 Invalid action）。""" - labels = ("允许", "Allow", "Authorize", "Approve", "授权") - # 1) DrissionPage 真实点击（与 device consent 同一套经验） - try: - candidates = [] - for sel in ("tag:button", "css:button", "css:[role='button']", "css:input[type='submit']"): - try: - found = page.eles(sel, timeout=0.3) or [] - except Exception: - found = [] - for el in found: - try: - t = (getattr(el, "text", None) or el.raw_text or "").strip() - except Exception: - t = "" - if not t: - continue - compact = re.sub(r"\s+", "", t) - if compact in labels or t in labels: - candidates.append((0, el, t)) - elif any(x in compact for x in labels) and "全部" not in compact and "All" not in compact: - candidates.append((1, el, t)) - candidates.sort(key=lambda x: x[0]) - for _, el, t in candidates: - try: - el.click() # real click - log(f"browser REAL click allow: {t!r}") - return True - except Exception as exc: # noqa: BLE001 - log(f"real click {t!r} failed: {exc}") - try: - el.click(by_js=True) - log(f"browser JS click allow: {t!r}") - return True - except Exception: - continue - except Exception as exc: # noqa: BLE001 - log(f"ele allow click failed: {exc}") - - # 2) 表单 action=allow 后 submit（device consent 兜底同款） - try: - ret = _page_eval( - page, - r""" -() => { - const labels = new Set(['允许','Allow','Authorize','Approve','授权']); - const f = document.querySelector('form'); - if (f) { - let a = f.querySelector('input[name=action]'); - if (!a) { - a = document.createElement('input'); - a.type = 'hidden'; - a.name = 'action'; - f.appendChild(a); - } - a.value = 'allow'; - const btn = [...f.querySelectorAll('button,[role=button],input[type=submit]')] - .find(b => labels.has(((b.innerText||b.value||'').trim()))); - if (btn) { btn.click(); return {ok:true, via:'form-btn'}; } - f.submit(); - return {ok:true, via:'form-submit'}; - } - // 无 form：真实派发 pointer/mouse 事件 - const nodes = [...document.querySelectorAll('button,[role=button],input[type=submit],a')]; - const target = nodes.find(n => { - const t = ((n.innerText||n.textContent||n.value||'').replace(/\s+/g,'')).trim(); - return labels.has(t) || (t.includes('允许') && !t.includes('全部')); - }); - if (!target) { - return {ok:false, texts: nodes.slice(0,10).map(n => (n.innerText||n.value||'').trim()).filter(Boolean)}; - } - target.scrollIntoView({block:'center'}); - target.focus(); - for (const type of ['pointerdown','mousedown','pointerup','mouseup','click']) { - target.dispatchEvent(new MouseEvent(type, {bubbles:true, cancelable:true, view:window})); - } - return {ok:true, via:'mouse-events', text:(target.innerText||target.value||'').trim()}; -} -""", - ) - if isinstance(ret, dict) and ret.get("ok"): - log(f"browser allow fallback: {ret}") - return True - log(f"browser allow button not found: {ret!r}") - except Exception as exc: # noqa: BLE001 - log(f"click allow js failed: {exc}") - return False - - -def _browser_consent_server_action( - page: Any, - consent_url: str, - flow: AuthCodeFlow, - log: LogFn, -) -> str: - """在浏览器内 POST Next.js Server Action 批准 consent，直接拿 code。""" - html = _page_html(page) - action_id = _extract_next_action_id(html) - payload = [ - { - "action": "allow", - "clientId": CLIENT_ID, - "redirectUri": REDIRECT_URI, - "scope": SCOPE, - "state": flow.state, - "codeChallenge": flow.code_challenge, - "codeChallengeMethod": "S256", - "nonce": flow.nonce, - "principalType": "User", - "principalId": "", - "referrer": GROK_REFERRER, - } - ] - body = json.dumps(payload, separators=(",", ":")) - # 用 window key 轮询，兼容 DrissionPage 对 Promise 支持不一 - wrap = f""" -(() => {{ - const key = '__cpa_consent_' + Date.now(); - window[key] = null; - fetch({consent_url!r}, {{ - method: 'POST', - headers: {{ - 'Accept': 'text/x-component', - 'Content-Type': 'text/plain;charset=UTF-8', - 'Next-Action': {action_id!r}, - 'Origin': 'https://accounts.x.ai', - 'Referer': {consent_url!r}, - }}, - body: {body!r}, - credentials: 'include', - redirect: 'follow', - }}).then(async (r) => {{ - const text = await r.text(); - window[key] = {{status: r.status, url: r.url, text: text}}; - }}).catch((e) => {{ - window[key] = {{status: 0, url: '', text: String(e)}}; - }}); - return key; -}})() -""" - try: - key = _page_eval(page, wrap) - except Exception as exc: # noqa: BLE001 - raise OAuthCodeError(f"browser consent fetch 启动失败: {exc}") from exc - log(f"browser consent server-action post action_id={action_id[:12]}…") - deadline = time.time() + 20 - ret = None - while time.time() < deadline: - try: - ret = _page_eval(page, f"() => window[{key!r}]") - except Exception: - try: - ret = _page_eval(page, f"window[{key!r}]") - except Exception: - ret = None - if ret: - break - time.sleep(0.2) - if not isinstance(ret, dict): - raise OAuthCodeError(f"browser consent 无响应: {ret!r}") - status = int(ret.get("status") or 0) - text = str(ret.get("text") or "") - final = str(ret.get("url") or "") - if status and (status < 200 or status >= 300): - raise OAuthCodeError(f"browser consent HTTP {status}: {_short(text, 240)}") - if "code=" in final: - qs = parse_qs(urlparse(final).query) - code = (qs.get("code") or [""])[0].strip() - if code: - log("browser got code from consent redirect url") - return code - # 响应体 / RSC - try: - code = parse_consent_code(text) - if code: - log("browser got code from consent server-action body") - return code - except OAuthCodeError: - pass - # 有时 code 在 text 的 redirect 串里 - m = re.search(r"[?&]code=([A-Za-z0-9._~\-]+)", text) - if m: - log("browser got code from consent body regex") - return m.group(1) - raise OAuthCodeError(f"browser consent 未返回 code: {_short(text, 240)}") - - -def _browser_fetch_token(page: Any, code: str, flow: AuthCodeFlow, log: LogFn) -> TokenResult: - """在浏览器上下文用 fetch 换 token，绕开 Python TLS 对 auth.x.ai 的 EOF。""" - form = { - "grant_type": "authorization_code", - "code": code, - "redirect_uri": REDIRECT_URI, - "client_id": CLIENT_ID, - "code_verifier": flow.code_verifier, - } - body = urlencode(form) - js = r""" -(tokenUrl, body, ua, ver) => { - return fetch(tokenUrl, { - method: 'POST', - headers: { - 'Content-Type': 'application/x-www-form-urlencoded', - 'Accept': '*/*', - 'User-Agent': ua, - 'X-Grok-Client-Version': ver, - }, - body: body, - credentials: 'include', - }).then(async (r) => { - const text = await r.text(); - return {status: r.status, text: text}; - }).catch((e) => ({status: 0, text: String(e)})); -} -""" - # 先到 auth 域，减少跨站限制 - try: - page.get(ISSUER + "/") - time.sleep(0.3) - except Exception: - pass - ret = None - # DrissionPage 对 Promise 支持不一，做短轮询包装 - wrap = r""" -(tokenUrl, body, ua, ver) => { - const key = '__cpa_token_result_' + Date.now(); - window[key] = null; - fetch(tokenUrl, { - method: 'POST', - headers: { - 'Content-Type': 'application/x-www-form-urlencoded', - 'Accept': '*/*', - 'User-Agent': ua, - 'X-Grok-Client-Version': ver, - }, - body: body, - credentials: 'include', - }).then(async (r) => { - const text = await r.text(); - window[key] = {status: r.status, text: text}; - }).catch((e) => { - window[key] = {status: 0, text: String(e)}; - }); - return key; -} -""" - try: - key = _page_eval(page, wrap, TOKEN_URL, body, GROK_TOKEN_UA, GROK_VERSION) - except Exception: - # 无参回退：把参数内联 - key = _page_eval( - page, - f""" -(() => {{ - const key = '__cpa_token_result_' + Date.now(); - window[key] = null; - fetch({TOKEN_URL!r}, {{ - method: 'POST', - headers: {{ - 'Content-Type': 'application/x-www-form-urlencoded', - 'Accept': '*/*', - 'User-Agent': {GROK_TOKEN_UA!r}, - 'X-Grok-Client-Version': {GROK_VERSION!r}, - }}, - body: {body!r}, - credentials: 'include', - }}).then(async (r) => {{ - const text = await r.text(); - window[key] = {{status: r.status, text: text}}; - }}).catch((e) => {{ - window[key] = {{status: 0, text: String(e)}}; - }}); - return key; -}})() -""", - ) - deadline = time.time() + 30 - while time.time() < deadline: - try: - ret = _page_eval(page, f"() => window[{key!r}]") - except Exception: - try: - ret = _page_eval(page, f"window[{key!r}]") - except Exception as exc: - raise OAuthCodeError(f"读取 browser token 结果失败: {exc}") from exc - if ret: - break - time.sleep(0.2) - if not isinstance(ret, dict): - raise OAuthCodeError(f"browser token 无响应: {ret!r}") - status = int(ret.get("status") or 0) - text = str(ret.get("text") or "") - if status < 200 or status >= 300: - raise OAuthCodeError(f"browser token HTTP {status}: {_short(text, 300)}") - try: - data = json.loads(text) - except Exception as e: - raise OAuthCodeError(f"browser token 非 JSON: {_short(text)}") from e - if not isinstance(data, dict) or not data.get("access_token"): - raise OAuthCodeError(f"browser token 缺少 access_token: {data!r}") - access = str(data["access_token"]).strip() - refresh = str(data.get("refresh_token") or "").strip() - if not refresh: - raise OAuthCodeError("browser token 缺少 refresh_token") - referrer = "" - try: - referrer = str(jwt_payload(access).get("referrer") or "") - except Exception: - pass - return TokenResult( - access_token=access, - refresh_token=refresh, - id_token=(str(data["id_token"]).strip() if data.get("id_token") else None), - token_type=str(data.get("token_type") or "Bearer"), - expires_in=int(data.get("expires_in") or 21600), - raw=data, - referrer=referrer, - ) - - -def mint_from_sso_browser( - sso_cookie: str, - page: Any, - *, - log: LogFn | None = None, - require_referrer: bool = True, - timeout_sec: float = 90.0, -) -> dict[str, Any]: - """用注册浏览器完成 SSO→PKCE（绕开 Python TLS 访问 auth.x.ai 失败）。 - - 流程： - 1. 写入 sso cookie - 2. 打开 authorize(referrer=grok-build) - 3. 在 consent 页点击允许 / 或解析 callback code - 4. 浏览器 fetch oauth2/token - """ - log = log or _noop_log - sso = normalize_sso_cookie(sso_cookie) - if not sso: - raise OAuthCodeError("sso cookie 为空") - if page is None: - raise OAuthCodeError("page 为空") - - flow = new_auth_code_flow() - params = { - "response_type": "code", - "client_id": CLIENT_ID, - "redirect_uri": REDIRECT_URI, - "scope": SCOPE, - "code_challenge": flow.code_challenge, - "code_challenge_method": "S256", - "state": flow.state, - "nonce": flow.nonce, - "referrer": GROK_REFERRER, - } - auth_url = f"{AUTHORIZE_URL}?{urlencode(params)}" - log(f"browser PKCE authorize referrer={GROK_REFERRER}") - _ensure_sso_on_page(page, sso, log) - - try: - page.get(auth_url) - except Exception as exc: # noqa: BLE001 - raise OAuthCodeError(f"browser 打开 authorize 失败: {exc}") from exc - - code = "" - deadline = time.time() + max(30.0, float(timeout_sec)) - last_url = "" - consent_attempts = 0 - server_action_tried = False - wait_after_click_until = 0.0 - - def _code_from_url(u: str) -> str: - if not u or "code=" not in u: - return "" - if not any(x in u for x in ("127.0.0.1", "localhost", "callback", "redirect")): - # 仍允许任意带 code 的 redirect - if "code=" not in u: - return "" - qs = parse_qs(urlparse(u).query) - return (qs.get("code") or [""])[0].strip() - - while time.time() < deadline: - try: - url = str(getattr(page, "url", "") or "") - except Exception: - url = "" - if url and url != last_url: - log(f"browser url: {_short(url, 140)}") - last_url = url - - # callback 已跳到 redirect_uri?code= - code = _code_from_url(url) - if code: - log("browser got code from redirect") - break - - # consent 页：真实点击 Allow；点后短暂等待跳转；仍不行再 Server Action POST - if "/oauth2/consent" in url or "/consent" in url: - # 刚点过，先等 redirect - if wait_after_click_until and time.time() < wait_after_click_until: - time.sleep(0.3) - continue - - consent_attempts += 1 - if consent_attempts <= 4: - clicked = _browser_click_allow(page, log) - if clicked: - wait_after_click_until = time.time() + 4.0 - time.sleep(0.6) - # 点后立刻看 url / html - try: - url2 = str(getattr(page, "url", "") or "") - except Exception: - url2 = url - code = _code_from_url(url2) - if code: - log("browser got code after allow click") - break - html = _page_html(page) - if html: - try: - code = parse_consent_code(html) - if code: - log("browser got code from consent html after click") - break - except OAuthCodeError: - pass - continue - - # 真实点击无效：浏览器内 Server Action POST（与 HTTP approve 同 payload） - if not server_action_tried: - server_action_tried = True - try: - code = _browser_consent_server_action(page, url, flow, log) - if code: - break - except Exception as exc: # noqa: BLE001 - log(f"browser consent server-action failed: {exc}") - time.sleep(0.5) - continue - - if "sign-in" in url or "sign-up" in url: - # cookie 可能没带上，重写一次 - log("browser landed sign-in, re-inject sso") - _ensure_sso_on_page(page, sso, log) - try: - page.get(auth_url) - except Exception: - pass - time.sleep(0.8) - continue - - # 中间跳转页：稍等 - time.sleep(0.4) - - if not code: - # 最后再扫一次页面 HTML - try: - html = _page_html(page) - if html: - try: - code = parse_consent_code(html) - except OAuthCodeError: - code = "" - except Exception: - code = "" - if not code: - raise OAuthCodeError( - f"browser PKCE 超时未拿到 code（last_url={_short(last_url, 160)}）" - ) - - token = _browser_fetch_token(page, code, flow, log) - if token.referrer != GROK_REFERRER: - msg = f"access_token 未包含预期 referrer（got={token.referrer!r}）" - if require_referrer: - raise OAuthCodeError(msg) - log(f"WARN {msg}") - else: - log("browser access_token referrer=grok-build ok") - log(f"browser token ok expires_in={token.expires_in}") - return { - "access_token": token.access_token, - "refresh_token": token.refresh_token, - "id_token": token.id_token, - "token_type": token.token_type, - "expires_in": token.expires_in, - "referrer": token.referrer, - "sso": sso, - }