chaos/grok-keygen-new
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert mint path to pre-browser PKCE baseline (2e833f2).

Browse Source 
Restore HTTP-only SSO OAuth with curl_cffi then std requests fallback;
remove browser PKCE prefer/fallback knobs and mint_from_sso_browser.
This commit is contained in:
chaos committed 2026-07-14 14:06:13 +08:00
1 parent d74d14d3f4
commit 4abdc8aa4a
5 files changed
+51 -751
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

oidc_mint/oauth_code.py
+30 -670
View File
@@ -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,
}