Git-creat7
bb2ef84c03
- cpa_remote_url 示例改为「你的CPA地址」 - cpa_auth_dir 示例改为「你的CPA auth目录」 - 删除 WSL / 跨机器挂载相关表述
534 lines
18 KiB
Python
534 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
SSO cookie → ~/.grok/auth.json 格式（纯 HTTP Device Flow）
|
||
|
||
用法:
|
||
# 单个 / 批量 SSO，写出多个独立 auth 文件（每个可直接 cp 到 ~/.grok/auth.json）
|
||
python3 sso_to_auth_json.py --sso sso_list.txt --out-dir ./auth_out
|
||
|
||
# 合并到一个 json（key 带 user_id 后缀，避免覆盖）
|
||
python3 sso_to_auth_json.py --sso sso_list.txt --out auth_merged.json --merge
|
||
|
||
# 单行 sso
|
||
python3 sso_to_auth_json.py --sso-cookie 'eyJ...' --out ~/.grok/auth.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import secrets
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from curl_cffi import requests
|
||
|
||
CLIENT_ID = "b1a00492-073a-47ea-816f-4c329264a828"
|
||
OIDC_ISSUER = "https://auth.x.ai"
|
||
AUTH_KEY = f"{OIDC_ISSUER}::{CLIENT_ID}"
|
||
SCOPES = (
|
||
"openid profile email offline_access grok-cli:access "
|
||
"api:access conversations:read conversations:write"
|
||
)
|
||
|
||
# --- CLIProxyAPI (CPA) 扁平格式常量 ------------------------------------------
|
||
# CPA 的 internal/auth/xai/token.go TokenStorage 读的是扁平字段。
|
||
# Build/CLI token（scope 含 grok-cli:access）必须走 cli-chat-proxy.grok.com，
|
||
# 不能用默认 api.x.ai/v1（那是计费通道，会 402）。
|
||
CPA_TOKEN_ENDPOINT = f"{OIDC_ISSUER}/oauth2/token"
|
||
CPA_GROK_BASE_URL = "https://cli-chat-proxy.grok.com/v1"
|
||
CPA_GROK_HEADERS = {
|
||
"X-XAI-Token-Auth": "xai-grok-cli",
|
||
"x-grok-client-version": "0.2.93",
|
||
"x-grok-client-identifier": "grok-shell",
|
||
}
|
||
|
||
|
||
def b64url_decode(seg: str) -> bytes:
|
||
seg += "=" * (-len(seg) % 4)
|
||
return base64.urlsafe_b64decode(seg)
|
||
|
||
|
||
def decode_jwt_payload(token: str) -> dict:
|
||
try:
|
||
return json.loads(b64url_decode(token.split(".")[1]))
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def rfc3339_ns(ts: float | None = None) -> str:
|
||
"""2026-07-10T01:00:00.000000000Z"""
|
||
if ts is None:
|
||
ts = time.time()
|
||
dt = datetime.fromtimestamp(ts, tz=timezone.utc)
|
||
return dt.strftime("%Y-%m-%dT%H:%M:%S") + ".000000000Z"
|
||
|
||
|
||
def _urlopen(req, proxy: str = "", timeout: int = 15):
|
||
"""urllib 请求，proxy 非空时走代理。"""
|
||
if proxy:
|
||
opener = urllib.request.build_opener(
|
||
urllib.request.ProxyHandler({"http": proxy, "https": proxy})
|
||
)
|
||
return opener.open(req, timeout=timeout)
|
||
return urllib.request.urlopen(req, timeout=timeout)
|
||
|
||
|
||
def request_device_code(proxy: str = "", log=print) -> dict | None:
|
||
data = urllib.parse.urlencode({"client_id": CLIENT_ID, "scope": SCOPES}).encode()
|
||
req = urllib.request.Request(
|
||
f"{OIDC_ISSUER}/oauth2/device/code",
|
||
data=data,
|
||
method="POST",
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
try:
|
||
with _urlopen(req, proxy=proxy, timeout=15) as resp:
|
||
return json.loads(resp.read())
|
||
except urllib.error.HTTPError as e:
|
||
log(f" ❌ device/code HTTP {e.code}: {e.read().decode()[:200]}")
|
||
return None
|
||
|
||
|
||
def poll_token(device_code: str, interval: int, expires_in: int, timeout: int = 60, proxy: str = "", log=print) -> dict | None:
|
||
deadline = time.time() + min(expires_in, timeout)
|
||
while time.time() < deadline:
|
||
time.sleep(interval)
|
||
data = urllib.parse.urlencode(
|
||
{
|
||
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
|
||
"client_id": CLIENT_ID,
|
||
"device_code": device_code,
|
||
}
|
||
).encode()
|
||
req = urllib.request.Request(
|
||
f"{OIDC_ISSUER}/oauth2/token",
|
||
data=data,
|
||
method="POST",
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
try:
|
||
with _urlopen(req, proxy=proxy, timeout=15) as resp:
|
||
return json.loads(resp.read())
|
||
except urllib.error.HTTPError as e:
|
||
err = json.loads(e.read())
|
||
error = err.get("error", "")
|
||
if error == "authorization_pending":
|
||
continue
|
||
if error == "slow_down":
|
||
interval += 5
|
||
continue
|
||
log(f" ❌ token: {error}")
|
||
return None
|
||
log(" ❌ 轮询超时")
|
||
return None
|
||
|
||
|
||
def sso_to_token(sso_cookie: str, proxy: str = "", log=print) -> dict | None:
|
||
"""SSO cookie → token dict (access/refresh/expires_in)。proxy 非空时全程走代理。"""
|
||
proxies = {"http": proxy, "https": proxy} if proxy else None
|
||
s = requests.Session()
|
||
if proxies:
|
||
s.proxies = proxies
|
||
s.cookies.set("sso", sso_cookie, domain=".x.ai")
|
||
|
||
try:
|
||
r = s.get("https://accounts.x.ai/", impersonate="chrome", timeout=15)
|
||
except Exception as e:
|
||
log(f" ❌ 网络错误: {e}")
|
||
return None
|
||
if "sign-in" in r.url or "sign-up" in r.url:
|
||
log(" ❌ sso 无效")
|
||
return None
|
||
log(" ✅ sso 有效")
|
||
|
||
log(" 🔑 Device Flow...")
|
||
dc = request_device_code(proxy=proxy, log=log)
|
||
if not dc:
|
||
return None
|
||
log(f" 📋 user_code: {dc.get('user_code')}")
|
||
|
||
try:
|
||
s.get(dc["verification_uri_complete"], impersonate="chrome", timeout=15)
|
||
r = s.post(
|
||
f"{OIDC_ISSUER}/oauth2/device/verify",
|
||
data={"user_code": dc["user_code"]},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
impersonate="chrome",
|
||
timeout=15,
|
||
allow_redirects=True,
|
||
)
|
||
if "consent" not in r.url:
|
||
log(f" ❌ verify 失败: {r.url}")
|
||
return None
|
||
except Exception as e:
|
||
log(f" ❌ verify 异常: {e}")
|
||
return None
|
||
|
||
try:
|
||
r = s.post(
|
||
f"{OIDC_ISSUER}/oauth2/device/approve",
|
||
data={
|
||
"user_code": dc["user_code"],
|
||
"action": "allow",
|
||
"principal_type": "User",
|
||
"principal_id": "",
|
||
},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
impersonate="chrome",
|
||
timeout=15,
|
||
allow_redirects=True,
|
||
)
|
||
if "done" not in r.url:
|
||
log(f" ❌ approve 失败: {r.url}")
|
||
return None
|
||
log(" ✅ 授权确认")
|
||
except Exception as e:
|
||
log(f" ❌ approve 异常: {e}")
|
||
return None
|
||
|
||
token = poll_token(
|
||
dc["device_code"],
|
||
dc.get("interval", 5),
|
||
dc.get("expires_in", 1800),
|
||
proxy=proxy,
|
||
log=log,
|
||
)
|
||
if not token:
|
||
return None
|
||
log(
|
||
f" ✅ access_token (expires_in={token.get('expires_in')}s)"
|
||
+ (" + refresh_token" if token.get("refresh_token") else "")
|
||
)
|
||
return token
|
||
|
||
|
||
def token_to_auth_entry(token: dict, email: str = "") -> tuple[str, dict]:
|
||
"""
|
||
返回 (top_level_key, entry)
|
||
top_level_key 固定为 issuer::client_id（与 ~/.grok/auth.json 一致）
|
||
"""
|
||
access = token.get("access_token") or token.get("key") or ""
|
||
refresh = token.get("refresh_token") or ""
|
||
payload = decode_jwt_payload(access)
|
||
|
||
user_id = payload.get("sub") or payload.get("principal_id") or ""
|
||
principal_id = payload.get("principal_id") or user_id
|
||
principal_type = payload.get("principal_type") or "User"
|
||
|
||
expires_in = int(token.get("expires_in") or 21600)
|
||
# 优先用 JWT exp
|
||
if "exp" in payload:
|
||
expires_at = rfc3339_ns(float(payload["exp"]))
|
||
else:
|
||
expires_at = rfc3339_ns(time.time() + expires_in)
|
||
|
||
iat = payload.get("iat")
|
||
create_time = rfc3339_ns(float(iat) if iat else time.time())
|
||
|
||
entry = {
|
||
"key": access,
|
||
"auth_mode": "oidc",
|
||
"create_time": create_time,
|
||
"user_id": user_id,
|
||
"email": email or "",
|
||
"principal_type": principal_type,
|
||
"principal_id": principal_id,
|
||
"refresh_token": refresh,
|
||
"expires_at": expires_at,
|
||
"oidc_issuer": OIDC_ISSUER,
|
||
"oidc_client_id": CLIENT_ID,
|
||
}
|
||
return AUTH_KEY, entry
|
||
|
||
|
||
def _iso_utc_from_unix(ts) -> str:
|
||
"""unix 秒 → CPA 认的 RFC3339（秒级，带 Z）。"""
|
||
try:
|
||
return datetime.fromtimestamp(int(ts), tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _safe_email_for_filename(email: str) -> str:
|
||
safe = "".join(ch if ch.isalnum() or ch in "._-@" else "_" for ch in email)
|
||
return safe or "unknown"
|
||
|
||
|
||
def token_to_cpa_record(token: dict, email: str = "") -> dict:
|
||
"""token dict → CLIProxyAPI 扁平 xai auth 记录。
|
||
|
||
对齐 CPA internal/auth/xai/token.go 的 TokenStorage 字段，以及
|
||
grok-build-auth build_cliproxyapi_auth_record 的输出。
|
||
"""
|
||
access = token.get("access_token") or token.get("key") or ""
|
||
refresh = token.get("refresh_token") or ""
|
||
id_token = token.get("id_token") or ""
|
||
payload = decode_jwt_payload(access)
|
||
id_payload = decode_jwt_payload(id_token) if id_token else {}
|
||
|
||
if not email:
|
||
email = id_payload.get("email") or payload.get("email") or ""
|
||
sub = payload.get("sub") or id_payload.get("sub") or ""
|
||
|
||
# expired: 优先 access token 的 exp，其次 expires_in 推算
|
||
expired = ""
|
||
if "exp" in payload:
|
||
expired = _iso_utc_from_unix(payload["exp"])
|
||
elif token.get("expires_in") is not None:
|
||
try:
|
||
expired = _iso_utc_from_unix(int(time.time()) + int(token["expires_in"]))
|
||
except Exception:
|
||
expired = ""
|
||
|
||
return {
|
||
"type": "xai",
|
||
"auth_kind": "oauth",
|
||
"email": email or "",
|
||
"sub": sub,
|
||
"access_token": access,
|
||
"refresh_token": refresh,
|
||
"id_token": id_token,
|
||
"token_type": token.get("token_type", "Bearer"),
|
||
"expires_in": token.get("expires_in", None),
|
||
"expired": expired,
|
||
"last_refresh": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||
"redirect_uri": "",
|
||
"token_endpoint": CPA_TOKEN_ENDPOINT,
|
||
"base_url": CPA_GROK_BASE_URL,
|
||
"disabled": False,
|
||
"headers": dict(CPA_GROK_HEADERS),
|
||
}
|
||
|
||
|
||
def cpa_auth_filename(record: dict) -> str:
|
||
"""生成 CPA auth 文件名：xai-<email>.json。"""
|
||
ident = str(record.get("email") or "").strip() or str(record.get("sub") or "").strip()
|
||
safe = _safe_email_for_filename(ident)
|
||
# 避免 email 本地部分已是 xai 时出现 "xai-xai..."
|
||
fname = safe if safe.lower().startswith("xai") else f"xai-{safe}"
|
||
return f"{fname}.json"
|
||
|
||
|
||
def write_cpa_auth(auth_dir: Path, record: dict) -> Path:
|
||
"""写出 CPA 可热加载的 xai-<email>.json（原子替换）。
|
||
|
||
无 email 时用 sub(user_id) 命名，避免多个无 email 账号写成同一个
|
||
xai-unknown.json 互相覆盖。
|
||
"""
|
||
auth_dir.mkdir(parents=True, exist_ok=True)
|
||
path = auth_dir / cpa_auth_filename(record)
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
tmp.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
return path
|
||
|
||
|
||
def upload_cpa_auth_remote(
|
||
base_url: str,
|
||
management_key: str,
|
||
record: dict,
|
||
timeout: int = 30,
|
||
) -> str:
|
||
"""通过 CPA Management API 上传 auth 文件到远程实例。
|
||
|
||
POST /v0/management/auth-files?name=<file.json>
|
||
Header: Authorization: Bearer <management_key>
|
||
Body: raw JSON auth record
|
||
"""
|
||
import requests
|
||
|
||
base = str(base_url or "").strip().rstrip("/")
|
||
key = str(management_key or "").strip()
|
||
if not base:
|
||
raise ValueError("cpa_remote_url 为空")
|
||
if not key:
|
||
raise ValueError("cpa_management_key 为空")
|
||
|
||
name = cpa_auth_filename(record)
|
||
url = f"{base}/v0/management/auth-files"
|
||
resp = requests.post(
|
||
url,
|
||
params={"name": name},
|
||
headers={
|
||
"Authorization": f"Bearer {key}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
data=json.dumps(record, ensure_ascii=False).encode("utf-8"),
|
||
timeout=timeout,
|
||
)
|
||
if resp.status_code >= 400:
|
||
body = (resp.text or "").strip()
|
||
if len(body) > 300:
|
||
body = body[:300] + "..."
|
||
raise RuntimeError(f"远程上传失败 HTTP {resp.status_code}: {body or resp.reason}")
|
||
return name
|
||
|
||
|
||
def write_auth_json(path: Path, auth_key: str, entry: dict) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
data = {auth_key: entry}
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def merge_auth_json(path: Path, auth_key: str, entry: dict, unique: bool = True) -> None:
|
||
"""
|
||
合并写入。unique=True 时 key 变成 issuer::client_id::user_id，避免多账号互相覆盖。
|
||
"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
existing: dict = {}
|
||
if path.exists():
|
||
try:
|
||
existing = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
existing = {}
|
||
key = auth_key
|
||
if unique and entry.get("user_id"):
|
||
key = f"{auth_key}::{entry['user_id']}"
|
||
existing[key] = entry
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
tmp.write_text(json.dumps(existing, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def load_sso_list(path: str | None, single: str | None) -> list[str]:
|
||
if single:
|
||
return [single.strip()]
|
||
if not path:
|
||
return []
|
||
out = []
|
||
for line in Path(path).read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
# 兼容 邮箱----密码----sso
|
||
if "----" in line:
|
||
parts = line.split("----")
|
||
line = parts[-1].strip()
|
||
out.append(line)
|
||
return out
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="SSO cookie → grok auth.json (纯 HTTP)")
|
||
ap.add_argument("--sso", metavar="FILE", help="sso 列表文件（一行一个 JWT，或 邮箱----密码----sso）")
|
||
ap.add_argument("--sso-cookie", metavar="JWT", help="单个 sso cookie")
|
||
ap.add_argument("--out", default=None, help="输出 auth.json 路径（单账号或 --merge）")
|
||
ap.add_argument(
|
||
"--out-dir",
|
||
default=None,
|
||
help="批量时每个账号写一个 {user_id}.json（可直接 cp 到 ~/.grok/auth.json）",
|
||
)
|
||
ap.add_argument(
|
||
"--merge",
|
||
action="store_true",
|
||
help="合并到 --out，key 用 issuer::client_id::user_id",
|
||
)
|
||
ap.add_argument("--delay", type=int, default=0, help="每个间隔秒数")
|
||
ap.add_argument("--email", default="", help="写入 entry.email（可选）")
|
||
ap.add_argument(
|
||
"--cpa-auth-dir",
|
||
default=None,
|
||
help="额外写出 CLIProxyAPI 扁平格式 xai-<email>.json 到该目录（CPA 热加载）",
|
||
)
|
||
ap.add_argument(
|
||
"--cpa-remote-url",
|
||
default=None,
|
||
help="远程 CPA 地址，如 http://你的CPA地址:8317；配合 --cpa-management-key 通过 Management API 上传",
|
||
)
|
||
ap.add_argument(
|
||
"--cpa-management-key",
|
||
default=None,
|
||
help="远程 CPA 管理密钥（remote-management.secret-key 明文）",
|
||
)
|
||
ap.add_argument("--proxy", default="", help="device-flow 走代理，如 http://127.0.0.1:7890")
|
||
args = ap.parse_args()
|
||
|
||
cookies = load_sso_list(args.sso, args.sso_cookie)
|
||
if not cookies:
|
||
ap.error("需要 --sso 或 --sso-cookie")
|
||
|
||
if args.cpa_remote_url and not args.cpa_management_key:
|
||
ap.error("使用 --cpa-remote-url 时必须同时提供 --cpa-management-key")
|
||
if args.cpa_management_key and not args.cpa_remote_url:
|
||
ap.error("使用 --cpa-management-key 时必须同时提供 --cpa-remote-url")
|
||
|
||
if len(cookies) > 1 and not args.out_dir and not args.merge:
|
||
# 默认批量写目录
|
||
args.out_dir = args.out_dir or "./auth_out"
|
||
print(f"批量模式默认 --out-dir {args.out_dir}")
|
||
|
||
# 只指定 CPA 目标时不再默认写官方 ~/.grok/auth.json
|
||
if (
|
||
args.out is None
|
||
and args.out_dir is None
|
||
and not args.cpa_auth_dir
|
||
and not args.cpa_remote_url
|
||
and len(cookies) == 1
|
||
):
|
||
args.out = str(Path.home() / ".grok" / "auth.json")
|
||
|
||
print(f"🚀 SSO → auth.json: {len(cookies)} 个, delay={args.delay}s")
|
||
ok = 0
|
||
fail = 0
|
||
|
||
for i, sso in enumerate(cookies, 1):
|
||
print(f"\n{'=' * 60}\n[{i}/{len(cookies)}] ...\n{'=' * 60}")
|
||
try:
|
||
token = sso_to_token(sso, proxy=args.proxy)
|
||
if not token:
|
||
fail += 1
|
||
print(f" ❌ [{i}] 失败")
|
||
continue
|
||
key, entry = token_to_auth_entry(token, email=args.email)
|
||
uid = entry.get("user_id") or secrets.token_hex(4)
|
||
|
||
if args.out_dir:
|
||
p = Path(args.out_dir) / f"{uid}.json"
|
||
write_auth_json(p, key, entry)
|
||
print(f" 💾 {p}")
|
||
if args.out:
|
||
if args.merge or len(cookies) > 1:
|
||
merge_auth_json(Path(args.out), key, entry, unique=True)
|
||
print(f" 💾 merge → {args.out}")
|
||
else:
|
||
write_auth_json(Path(args.out), key, entry)
|
||
print(f" 💾 {args.out}")
|
||
|
||
if args.cpa_auth_dir or args.cpa_remote_url:
|
||
record = token_to_cpa_record(token, email=args.email)
|
||
if args.cpa_auth_dir:
|
||
cp = write_cpa_auth(Path(args.cpa_auth_dir), record)
|
||
print(f" 💾 CPA 本地 → {cp}")
|
||
if args.cpa_remote_url:
|
||
name = upload_cpa_auth_remote(
|
||
args.cpa_remote_url,
|
||
args.cpa_management_key,
|
||
record,
|
||
)
|
||
print(f" 💾 CPA 远程 → {args.cpa_remote_url.rstrip('/')}/.../{name}")
|
||
|
||
ok += 1
|
||
print(f" ✅ [{i}] 完成 user_id={uid[:12]}...")
|
||
except Exception as e:
|
||
fail += 1
|
||
print(f" ❌ [{i}] 异常: {e}")
|
||
|
||
if args.delay > 0 and i < len(cookies):
|
||
time.sleep(args.delay)
|
||
|
||
print(f"\n{'=' * 60}\n📊 完成: {ok}/{len(cookies)} 成功, {fail} 失败")
|
||
return 0 if fail == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|