chaos
ee151343e0
Switch CPA export to SSO→Authorization Code with referrer=grok-build, keep device-code as optional fallback, and capture new register/auth artifacts.
233 lines
8.0 KiB
Python
233 lines
8.0 KiB
Python
"""注册成功钩子：SSO → OAuth Authorization Code(PKCE, referrer=grok-build)
|
||
→ 写出 CPA (CLIProxyAPI) 的 xai-<email>.json → 推送到远端 CLIProxyAPI 导入。
|
||
|
||
- 本地写盘目录：config['cpa_auth_dir']（默认 ./cpa_auths）
|
||
- 远端推送：POST config['cpa_remote_base'] + /v0/management/auth-files?name=...
|
||
认证 X-Management-Key: config['cpa_remote_secret']
|
||
|
||
2026-07 起：设备码铸造的 token 缺 referrer=grok-build，cli-chat-proxy 不可用。
|
||
默认优先走 SSO cookie 的授权码流程；仅在无 sso 且允许时才回退设备码。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
_REG_DIR = Path(__file__).resolve().parent
|
||
_DEFAULT_OUT = _REG_DIR / "cpa_auths"
|
||
|
||
|
||
def _resolve_out_dir(cfg: dict) -> Path:
|
||
raw = str(cfg.get("cpa_auth_dir") or _DEFAULT_OUT).strip()
|
||
p = Path(raw).expanduser()
|
||
if not p.is_absolute():
|
||
p = (_REG_DIR / p).resolve()
|
||
return p
|
||
|
||
|
||
def _record_failure(out_dir: Path, email: str, reason: str) -> None:
|
||
try:
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
with open(out_dir / "cpa_auth_failed.txt", "a", encoding="utf-8") as f:
|
||
f.write(f"{email}----{reason}----{int(time.time())}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _resolve_proxy(cfg: dict) -> str | None:
|
||
from oidc_mint import resolve_proxy, set_runtime_proxy
|
||
|
||
proxy = (cfg.get("mint_proxy") or cfg.get("proxy") or "").strip()
|
||
if not proxy:
|
||
proxy = (
|
||
os.environ.get("https_proxy")
|
||
or os.environ.get("HTTPS_PROXY")
|
||
or os.environ.get("http_proxy")
|
||
or ""
|
||
).strip()
|
||
resolved = resolve_proxy(proxy or None)
|
||
set_runtime_proxy(resolved or None)
|
||
return resolved or None
|
||
|
||
|
||
def _mint_tokens(
|
||
*,
|
||
email: str,
|
||
password: str,
|
||
sso: str,
|
||
page: Any | None,
|
||
cfg: dict,
|
||
log: Callable[[str], None],
|
||
proxy: str | None,
|
||
) -> dict[str, Any]:
|
||
"""优先 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_device = bool(cfg.get("cpa_allow_device_fallback", False))
|
||
timeout = float(cfg.get("mint_timeout_sec", 300) or 300)
|
||
|
||
if prefer_sso and sso_token:
|
||
log("[cpa] 使用 SSO→OAuth(PKCE, referrer=grok-build)")
|
||
try:
|
||
return mint_from_sso(
|
||
sso_token,
|
||
proxy=proxy,
|
||
log=lambda m: log(f"[Debug] {m}"),
|
||
require_referrer=bool(cfg.get("cpa_require_referrer", True)),
|
||
)
|
||
except OAuthCodeError as exc:
|
||
log(f"[!] SSO→OAuth 失败: {exc}")
|
||
if not allow_device:
|
||
raise
|
||
log("[cpa] 回退设备码铸造（可能缺 referrer）")
|
||
except Exception as exc: # noqa: BLE001
|
||
log(f"[!] SSO→OAuth 异常: {exc}")
|
||
if not allow_device:
|
||
raise
|
||
log("[cpa] 回退设备码铸造（可能缺 referrer）")
|
||
|
||
if not allow_device and not sso_token:
|
||
raise RuntimeError("无 sso cookie，且已禁用设备码回退；无法铸造带 referrer 的 token")
|
||
|
||
# 设备码回退（旧路径，通常无 referrer）
|
||
from oidc_mint import mint_with_browser
|
||
|
||
if page is None and not (email and password):
|
||
raise RuntimeError("设备码回退需要 page 或 email/password")
|
||
|
||
log("[cpa] 使用设备码铸造（兼容路径）")
|
||
tokens = mint_with_browser(
|
||
email=email,
|
||
password=password,
|
||
page=page,
|
||
proxy=proxy,
|
||
browser_timeout_sec=timeout,
|
||
force_standalone=(page is None),
|
||
cookies=None,
|
||
poll_log=lambda m: log(f"[Debug] {m}"),
|
||
)
|
||
return tokens
|
||
|
||
|
||
# ── 主入口 ──
|
||
|
||
def export_cpa_for_account(
|
||
email: str,
|
||
password: str = "",
|
||
*,
|
||
page: Any | None = None,
|
||
sso: str = "",
|
||
config: dict | None = None,
|
||
log_callback: Callable[[str], None] | None = None,
|
||
) -> dict:
|
||
"""铸造 OIDC → 写本地 xai-<email>.json → 推送远端 CLIProxyAPI。
|
||
|
||
优先使用 sso cookie 走 Authorization Code + referrer=grok-build。
|
||
返回 {ok, email, path, pushed, push_status?, error?}。
|
||
"""
|
||
cfg = config or {}
|
||
log = log_callback or (lambda m: print(m, flush=True))
|
||
|
||
if not cfg.get("cpa_export_enabled", True):
|
||
log("[cpa] 已关闭导出，跳过")
|
||
return {"ok": False, "skipped": True, "reason": "disabled"}
|
||
email = (email or "").strip()
|
||
if not email:
|
||
return {"ok": False, "error": "缺少 email", "email": email}
|
||
|
||
import cpa
|
||
from oidc_mint.oauth_code import normalize_sso_cookie
|
||
|
||
out_dir = _resolve_out_dir(cfg)
|
||
proxy = _resolve_proxy(cfg)
|
||
base_url = cfg.get("cpa_base_url") or cpa.CLI_BASE_URL
|
||
sso_token = normalize_sso_cookie(sso or "")
|
||
|
||
try:
|
||
tokens = _mint_tokens(
|
||
email=email,
|
||
password=password or "",
|
||
sso=sso_token,
|
||
page=page,
|
||
cfg=cfg,
|
||
log=log,
|
||
proxy=proxy,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
log(f"[!] 铸造失败: {exc}")
|
||
_record_failure(out_dir, email, str(exc))
|
||
if cfg.get("mint_required", False):
|
||
raise
|
||
return {"ok": False, "error": str(exc), "email": email}
|
||
|
||
try:
|
||
payload = cpa.build_cpa_xai_auth(
|
||
email=email,
|
||
access_token=tokens["access_token"],
|
||
refresh_token=tokens["refresh_token"],
|
||
id_token=tokens.get("id_token"),
|
||
expires_in=tokens.get("expires_in"),
|
||
base_url=base_url,
|
||
sso=tokens.get("sso") or sso_token or None,
|
||
)
|
||
path = cpa.write_cpa_xai_auth(out_dir, payload)
|
||
filename = Path(path).name
|
||
except Exception as exc: # noqa: BLE001
|
||
log(f"[!] 写本地文件失败: {exc}")
|
||
_record_failure(out_dir, email, f"write: {exc}")
|
||
if cfg.get("mint_required", False):
|
||
raise
|
||
return {"ok": False, "error": str(exc), "email": email}
|
||
|
||
ref = payload.get("referrer") or tokens.get("referrer") or ""
|
||
log(f"[Debug] 已写本地: {path} referrer={ref or '(empty)'}")
|
||
result: dict[str, Any] = {
|
||
"ok": True,
|
||
"email": email,
|
||
"path": str(path),
|
||
"pushed": False,
|
||
"referrer": ref,
|
||
}
|
||
|
||
# 推送远端 CLIProxyAPI
|
||
if cfg.get("cpa_push_enabled", False):
|
||
remote_base = str(cfg.get("cpa_remote_base") or "").strip()
|
||
secret = str(cfg.get("cpa_remote_secret") or "").strip()
|
||
if not remote_base or not secret:
|
||
log("[!] 推送已开启但未配置 cpa_remote_base/cpa_remote_secret，跳过推送")
|
||
else:
|
||
push_proxy = str(cfg.get("cpa_push_proxy") or "").strip() or None
|
||
verify_tls = bool(cfg.get("cpa_remote_verify_tls", True))
|
||
try:
|
||
ok, status, text = cpa.push_auth_file(
|
||
remote_base=remote_base,
|
||
secret=secret,
|
||
filename=filename,
|
||
payload=payload,
|
||
proxy=push_proxy,
|
||
verify_tls=verify_tls,
|
||
)
|
||
result["push_status"] = status
|
||
if ok:
|
||
result["pushed"] = True
|
||
log(f"[Debug] 已推送远端: {remote_base} (HTTP {status})")
|
||
else:
|
||
result["push_error"] = text[:300]
|
||
log(f"[!] [cpa] 推送远端失败 HTTP {status}: {text[:200]}")
|
||
if cfg.get("cpa_push_required", False):
|
||
result["ok"] = False
|
||
result["error"] = f"push HTTP {status}: {text[:200]}"
|
||
except Exception as exc: # noqa: BLE001
|
||
log(f"[!] [cpa] 推送远端异常: {exc}")
|
||
result["push_error"] = str(exc)
|
||
if cfg.get("cpa_push_required", False):
|
||
result["ok"] = False
|
||
result["error"] = f"push: {exc}"
|
||
|
||
return result
|