feat: Grok 注册机 — 注册即入库 CLIProxyAPI (CPA)

- GUI / CLI 批量注册 Grok 账号
- SSO → device-flow OAuth → 本地 auth 目录或远程 Management API
- Cloudflare / DuckMail / YYDS 临时邮箱
- Ctrl+C 安全停止，清理阶段不刷 traceback

.gitattributes

@@ -0,0 +1,4 @@
+*.py text eol=lf
+*.md text eol=lf
+*.json text eol=lf
+*.txt text eol=lf

.gitignore

@@ -0,0 +1,33 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+.venv/
+venv/
+env/
+
+accounts_*.txt
+mail_credentials.txt
+*.txt
+!requirements.txt
+
+config.json
+config.local.json
+config.local.*
+*.bak
+
+# 真实账号凭证 / OAuth token 输出，禁止提交
+auth_out.json
+auth_out/
+auth_merged.json
+*.auth.json
+sso_list.txt
+token.json
+
+*.log
+.DS_Store
+.idea/
+.vscode/
+DrissionPage/
+autoPortData/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 AaronL725
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,253 @@
+<div align="center">
+
+[![Grok Register — 注册即入库 CLIProxyAPI](assets/banner.png)](https://github.com/Git-creat7/grokRegister-cpa)
+
+批量注册 Grok 账号，注册成功后自动把 OAuth 凭证写入 [CLIProxyAPI (CPA)](https://github.com/router-for-me/CLIProxyAPI)：支持本地 auth 目录热加载，也支持 Management API 远程上传。
+
+<p>
+  <a href="LICENSE"><img src="https://img.shields.io/badge/License-MIT-blue.svg" alt="License: MIT"></a>
+  <img src="https://img.shields.io/badge/Python-3.9%2B-3776AB.svg" alt="Python 3.9+">
+  <img src="https://img.shields.io/badge/Interface-GUI%20%2B%20CLI-success.svg" alt="GUI + CLI">
+  <img src="https://img.shields.io/badge/Output-CLIProxyAPI-orange.svg" alt="CLIProxyAPI">
+</p>
+
+</div>
+
+---
+
+> 仅用于自动化流程研究、测试环境验证和个人学习。请遵守目标网站服务条款、当地法律法规与第三方服务限制。
+
+## 核心流程
+
+```text
+打开注册页 → 创建临时邮箱 → 收验证码 → 填资料 / 过人机验证
+   → 拿到 SSO cookie → device-flow 换 OAuth token
+   → 本地写入 cpa_auth_dir  和/或  POST 远程 CPA Management API
+   → CPA 热加载，立即可用
+```
+
+## 功能
+
+- 注册成功后自动入库 CPA（本地目录 / 远程 Management API，可同时开）
+- GUI + CLI 两种运行方式（CLI 仍会打开浏览器完成注册页）
+- Chromium/Chrome 自动处理 Turnstile
+- DuckMail / YYDS / Cloudflare 临时邮箱
+- 注册后可选开启 NSFW
+- 页面卡住重试、验证码失败换邮箱、浏览器重启与内存清理
+- CLI：一次 `Ctrl+C` 安全停止，清理阶段不刷 traceback；再按一次强制中断
+
+## 环境要求
+
+- Python 3.9+
+- Google Chrome 或 Chromium
+- 可用的 [CLIProxyAPI](https://github.com/router-for-me/CLIProxyAPI)
+- 能访问注册页、临时邮箱 API、`auth.x.ai` 的网络（device-flow 换 token 需要）
+
+## 安装
+
+```bash
+git clone https://github.com/Git-creat7/grokRegister-cpa.git
+cd grokRegister-cpa
+pip install -r requirements.txt
+cp config.example.json config.json
+```
+
+编辑 `config.json` 后运行。
+
+## 配置
+
+| 配置项 | 说明 |
+| --- | --- |
+| `cpa_auto_add` | 是否开启 CPA 自动入库 |
+| `cpa_auth_dir` | 本地 CPA auth 目录；写入 `xai-<email>.json`，可留空 |
+| `cpa_remote_url` | 远程 CPA 地址，如 `http://127.0.0.1:8317` |
+| `cpa_management_key` | 远程 CPA 管理密钥（`remote-management.secret-key` 明文） |
+| `email_provider` | `duckmail` / `yyds` / `cloudflare` |
+| `register_count` | 目标注册数量 |
+| `proxy` | 代理；device-flow 换 token 也走此代理 |
+| `enable_nsfw` | 注册后是否尝试开启 NSFW |
+| `cloudflare_api_base` | Cloudflare 临时邮箱 API 根地址 |
+| `cloudflare_api_key` | 默认匿名模式留空；admin 模式填 `ADMIN_PASSWORD` |
+| `cloudflare_auth_mode` | `none` / `bearer` / `x-api-key` / `x-admin-auth` / `query-key` |
+| `cloudflare_custom_auth` | Worker 全局密码（`PASSWORDS`），注入 `x-custom-auth` |
+| `cloudflare_path_*` | domains / accounts / token / messages 路径 |
+| `defaultDomains` | Cloudflare 默认收信域名 |
+
+### Cloudflare 邮箱（默认匿名）
+
+```json
+{
+  "email_provider": "cloudflare",
+  "cloudflare_api_base": "https://你的-worker-api-域名",
+  "cloudflare_api_key": "",
+  "cloudflare_auth_mode": "none",
+  "cloudflare_path_domains": "/api/domains",
+  "cloudflare_path_accounts": "/api/new_address",
+  "cloudflare_path_token": "/api/token",
+  "cloudflare_path_messages": "/api/mails",
+  "defaultDomains": "你的收信域名.com"
+}
+```
+
+匿名创建失败（例如 Turnstile）时可改 admin 创建：
+
+```json
+{
+  "cloudflare_api_key": "你的 ADMIN_PASSWORD",
+  "cloudflare_auth_mode": "x-admin-auth",
+  "cloudflare_path_accounts": "/admin/new_address"
+}
+```
+
+调试创建接口：
+
+```bash
+python cf_mail_debug.py \
+  --api-base "https://你的-worker-api-域名" \
+  --auth-mode x-admin-auth \
+  --api-key "你的 ADMIN_PASSWORD" \
+  --create-path /admin/new_address \
+  --domain "你的收信域名.com"
+```
+
+Worker 若配置了全局 `PASSWORDS`，再加：
+
+```json
+{ "cloudflare_custom_auth": "你的全局访问密码" }
+```
+
+## CPA 自动入库
+
+SSO 不是 CPA 凭据。程序会：
+
+1. 用 SSO 走 device-flow 向 `auth.x.ai` 换 `access_token` / `refresh_token`
+2. 组装 `type=xai` 扁平 auth（`cli-chat-proxy.grok.com`）
+3. 本地：`cpa_auth_dir` → `xai-<email>.json`（CPA 热加载）
+4. 远程：`POST {cpa_remote_url}/v0/management/auth-files?name=...`（需管理密钥）
+
+### 本地目录
+
+```json
+{
+  "cpa_auto_add": true,
+  "cpa_auth_dir": "/path/to/CLIProxyAPI/auths"
+}
+```
+
+跨机器 / WSL 可写挂载路径，例如  
+`//wsl.localhost/Ubuntu/home/you/CLIProxyAPI/auths`。
+
+### 远程 Management API
+
+```json
+{
+  "cpa_auto_add": true,
+  "cpa_auth_dir": "",
+  "cpa_remote_url": "http://127.0.0.1:8317",
+  "cpa_management_key": "你的管理密钥明文"
+}
+```
+
+要求 CPA：`remote-management.allow-remote` 按访问方式配置；密钥为配置里的明文（启动后配置文件可能被写成 bcrypt，上传仍用明文）。
+
+本地与远程可同时开启。日志前缀：`[CPA]`。
+
+### 独立转换
+
+已有 SSO 时可脱离注册流程：
+
+```bash
+# 写本地目录
+python sso_to_auth_json.py --sso sso_list.txt --cpa-auth-dir /path/to/auths
+
+# 上传远程 CPA
+python sso_to_auth_json.py --sso sso_list.txt \
+  --cpa-remote-url http://127.0.0.1:8317 \
+  --cpa-management-key '你的管理密钥'
+
+# 单个 cookie + 代理
+python sso_to_auth_json.py --sso-cookie 'eyJ...' \
+  --cpa-auth-dir ./auths \
+  --proxy http://127.0.0.1:7890
+```
+
+`sso_list.txt`：一行一个 SSO，或 `邮箱----密码----sso`。
+
+## 运行
+
+### CLI
+
+```bash
+python grok_register_ttk.py cli
+```
+
+提示后输入 `start`。  
+`Ctrl+C` 一次：当前账号收尾后停止；清理浏览器时不会因二次中断刷 traceback。再按一次强制退出。
+
+### GUI
+
+```bash
+python grok_register_ttk.py
+```
+
+可在界面里改 CPA 开关、auth 目录、远程地址与管理密钥。
+
+## 输出文件
+
+| 文件 | 内容 |
+| --- | --- |
+| `accounts_*.txt` | 邮箱、密码、SSO |
+| `mail_credentials.txt` | 临时邮箱凭证 |
+
+均含敏感信息，已在 `.gitignore` 中忽略。`config.json` 也不提交，请用 `config.example.json` 复制。
+
+## 稳定性
+
+- 每账号结束后重启浏览器
+- 每成功 5 个账号做一次内存清理
+- 邮箱提交后确认页面前进，避免空等验证码
+- 未收到验证码时换邮箱重试
+- 最终页卡住时重试当前账号
+
+## 常见问题
+
+**CPA 没出现新账号**  
+检查 `cpa_auto_add`、`cpa_auth_dir` 或 `cpa_remote_url` + `cpa_management_key`；看 `[CPA]` 日志是否换 token / 上传成功；本机/服务器能否访问 `auth.x.ai`。
+
+**远程上传失败**  
+确认 CPA 管理 API 已启用、密钥明文正确；远程访问需 `allow-remote: true`。可用：
+
+```bash
+curl -H "Authorization: Bearer <管理密钥>" \
+  http://127.0.0.1:8317/v0/management/auth-files
+```
+
+**CLI 为什么还开浏览器**  
+CLI 只是不启动 Tk；注册页、Turnstile、SSO 仍依赖真实浏览器。
+
+**NSFW 失败**  
+常见为 Cloudflare 拦截。账号仍会保存并入库 CPA。
+
+**国内服务器调模型超时**  
+入库成功只说明凭证到了 CPA；调用上游 `cli-chat-proxy.grok.com` 还需服务器出网可达（或配置 CPA `proxy-url`）。
+
+## 目录结构
+
+```text
+.
+├── grok_register_ttk.py      # 主程序（GUI / CLI + CPA 入库）
+├── sso_to_auth_json.py       # SSO → CPA 转换（可独立运行）
+├── cf_mail_debug.py          # Cloudflare 邮箱调试
+├── config.example.json
+├── requirements.txt
+├── tests/
+└── assets/banner.png
+```
+
+## License
+
+[MIT](LICENSE)
+
+## Acknowledgments
+
+Thanks to [linux.do](https://linux.do) and [CLIProxyAPI](https://github.com/router-for-me/CLIProxyAPI).

cf_mail_debug.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import argparse
+import re
+import secrets
+import string
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+from curl_cffi import requests
+
+
+def extract_code(text: str, subject: str = "") -> Optional[str]:
+    if subject:
+        m = re.search(r"\b([A-Z0-9]{3}-[A-Z0-9]{3})\b", subject, re.IGNORECASE)
+        if m:
+            return m.group(1)
+    m = re.search(r"\b([A-Z0-9]{3}-[A-Z0-9]{3})\b", text, re.IGNORECASE)
+    if m:
+        return m.group(1)
+    for p in [
+        r"verification\s+code[:\s]+(\d{4,8})",
+        r"your\s+code[:\s]+(\d{4,8})",
+        r"confirm(?:ation)?\s+code[:\s]+(\d{4,8})",
+    ]:
+        m = re.search(p, text, re.IGNORECASE)
+        if m:
+            return m.group(1)
+    return None
+
+
+def json_or_text(resp: requests.Response) -> Tuple[Optional[Dict[str, Any]], str]:
+    try:
+        data = resp.json()
+        return data, ""
+    except Exception:
+        return None, (resp.text or "")[:400]
+
+
+def generate_username(length: int = 10) -> str:
+    """生成 cloudflare_temp_email admin API 需要的随机邮箱名称。"""
+    chars = string.ascii_lowercase + string.digits
+    return "".join(secrets.choice(chars) for _ in range(length))
+
+
+def normalize_path(path: str, default_path: str) -> str:
+    """标准化 API 路径，避免漏写开头斜杠。"""
+    raw = (path or default_path).strip() or default_path
+    return raw if raw.startswith("/") else f"/{raw}"
+
+
+def build_auth_headers(auth_mode: str, api_key: str, content_type: bool = False) -> Dict[str, str]:
+    """按调试参数构造 Cloudflare 临时邮箱接口鉴权请求头。"""
+    headers = {"Content-Type": "application/json"} if content_type else {}
+    key = (api_key or "").strip()
+    mode = (auth_mode or "none").strip().lower()
+    if not key:
+        return headers
+    if mode == "x-admin-auth":
+        headers["x-admin-auth"] = key
+    elif mode == "x-api-key":
+        headers["X-API-Key"] = key
+    elif mode == "bearer":
+        headers["Authorization"] = f"Bearer {key}"
+    return headers
+
+
+def create_address(
+    api_base: str,
+    auth_mode: str = "none",
+    api_key: str = "",
+    create_path: str = "/api/new_address",
+    domain: str = "",
+    name: str = "",
+) -> Tuple[str, str]:
+    """创建 Cloudflare 临时邮箱地址，支持匿名 API 和 admin API。"""
+    path = normalize_path(create_path, "/api/new_address")
+    is_admin_create = path.rstrip("/").lower() == "/admin/new_address"
+    if is_admin_create:
+        payload: Dict[str, Any] = {
+            "name": name.strip() if name.strip() else generate_username(),
+            "enablePrefix": True,
+        }
+        if domain.strip():
+            payload["domain"] = domain.strip()
+        headers = build_auth_headers(auth_mode, api_key, content_type=True)
+    else:
+        payload = {}
+        if domain.strip():
+            payload["domain"] = domain.strip()
+        headers = {"Content-Type": "application/json"}
+    resp = requests.post(
+        f"{api_base.rstrip('/')}{path}",
+        json=payload,
+        headers=headers,
+        timeout=20,
+    )
+    resp.raise_for_status()
+    data, raw = json_or_text(resp)
+    if not data:
+        raise RuntimeError(f"{path} 非JSON: {raw}")
+    address = str(data.get("address", "")).strip()
+    jwt = str(data.get("jwt", "")).strip()
+    if not address or not jwt:
+        raise RuntimeError(f"{path} 缺少 address/jwt: {data}")
+    return address, jwt
+
+
+def fetch_box(api_base: str, jwt: str, path: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
+    resp = requests.get(
+        f"{api_base.rstrip('/')}{path}",
+        params=params,
+        headers={"Authorization": f"Bearer {jwt}"},
+        timeout=20,
+    )
+    if resp.status_code >= 400:
+        return []
+    data, _ = json_or_text(resp)
+    if not isinstance(data, dict):
+        return []
+    if isinstance(data.get("results"), list):
+        return data["results"]
+    if isinstance(data.get("data"), list):
+        return data["data"]
+    if isinstance(data.get("messages"), list):
+        return data["messages"]
+    return []
+
+
+def probe_all_boxes(api_base: str, jwt: str) -> List[Tuple[str, List[Dict[str, Any]]]]:
+    probes = [
+        ("/api/mails", {"limit": 20, "offset": 0}),
+        ("/api/sendbox", {"limit": 20, "offset": 0}),
+        ("/api/mails", {"limit": 20, "offset": 0, "box": "trash"}),
+        ("/api/mails", {"limit": 20, "offset": 0, "folder": "trash"}),
+        ("/api/mails", {"limit": 20, "offset": 0, "deleted": "1"}),
+        ("/api/mails", {"limit": 20, "offset": 0, "status": "deleted"}),
+    ]
+    out: List[Tuple[str, List[Dict[str, Any]]]] = []
+    for path, params in probes:
+        mails = fetch_box(api_base, jwt, path, params)
+        out.append((f"{path}?{params}", mails))
+    return out
+
+
+def get_detail(api_base: str, jwt: str, mail_id: Any) -> Dict[str, Any]:
+    for url in [
+        f"{api_base.rstrip('/')}/api/mail/{mail_id}",
+        f"{api_base.rstrip('/')}/api/mails/{mail_id}",
+    ]:
+        try:
+            resp = requests.get(url, headers={"Authorization": f"Bearer {jwt}"}, timeout=20)
+            if resp.status_code >= 400:
+                continue
+            data, _ = json_or_text(resp)
+            if isinstance(data, dict):
+                return data
+        except Exception:
+            continue
+    return {}
+
+
+def flatten_mail_text(item: Dict[str, Any], detail: Dict[str, Any]) -> Tuple[str, str]:
+    subject = str(item.get("subject") or detail.get("subject") or "")
+    parts: List[str] = []
+    for src in (item, detail):
+        for k in ("text", "raw", "content", "intro", "body", "snippet"):
+            v = src.get(k)
+            if isinstance(v, str) and v.strip():
+                parts.append(v)
+        html_val = src.get("html")
+        if isinstance(html_val, str):
+            html_val = [html_val]
+        if isinstance(html_val, list):
+            for h in html_val:
+                if isinstance(h, str):
+                    parts.append(re.sub(r"<[^>]+>", " ", h))
+    return subject, "\n".join(parts)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--api-base", required=True)
+    ap.add_argument("--address", default="")
+    ap.add_argument("--credential", default="")
+    ap.add_argument("--auth-mode", default="none", choices=["none", "bearer", "x-api-key", "x-admin-auth"])
+    ap.add_argument("--api-key", default="")
+    ap.add_argument("--create-path", default="/api/new_address")
+    ap.add_argument("--domain", default="")
+    ap.add_argument("--name", default="")
+    ap.add_argument("--timeout", type=int, default=180)
+    ap.add_argument("--interval", type=int, default=3)
+    args = ap.parse_args()
+
+    address = args.address.strip()
+    credential = args.credential.strip()
+    if not credential:
+        address, credential = create_address(
+            args.api_base,
+            auth_mode=args.auth_mode,
+            api_key=args.api_key,
+            create_path=args.create_path,
+            domain=args.domain,
+            name=args.name,
+        )
+        print(f"[NEW] address={address}")
+        print(f"[NEW] credential(jwt)={credential}")
+    else:
+        print(f"[USE] address={address or '(unknown, from credential)'}")
+
+    deadline = time.time() + max(args.timeout, 1)
+    seen_ids = set()
+    while time.time() < deadline:
+        boxes = probe_all_boxes(args.api_base, credential)
+        total = 0
+        for name, mails in boxes:
+            if mails:
+                print(f"[BOX] {name} -> {len(mails)}")
+            total += len(mails)
+            for m in mails:
+                mail_id = m.get("id") or m.get("mail_id")
+                if not mail_id or mail_id in seen_ids:
+                    continue
+                seen_ids.add(mail_id)
+                detail = get_detail(args.api_base, credential, mail_id)
+                subj, text = flatten_mail_text(m, detail)
+                code = extract_code(text, subj)
+                print(f"[MAIL] id={mail_id} subject={subj!r} code={code!r}")
+                if code:
+                    print(f"[FOUND] {code}")
+                    return
+        if total == 0:
+            print("[INFO] no mails yet")
+        time.sleep(max(args.interval, 1))
+    print("[TIMEOUT] no code found")
+
+
+if __name__ == "__main__":
+    main()

config.example.json

@@ -0,0 +1,23 @@
+{
+  "duckmail_api_key": "",
+  "cloudflare_api_base": "https://temp-mail.example.com",
+  "cloudflare_api_key": "",
+  "cloudflare_auth_mode": "none",
+  "cloudflare_custom_auth": "",
+  "cloudflare_path_domains": "/api/domains",
+  "cloudflare_path_accounts": "/api/new_address",
+  "cloudflare_path_token": "/api/token",
+  "cloudflare_path_messages": "/api/mails",
+  "proxy": "",
+  "enable_nsfw": true,
+  "register_count": 1,
+  "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",
+  "cpa_auto_add": false,
+  "cpa_auth_dir": "",
+  "cpa_remote_url": "",
+  "cpa_management_key": "",
+  "defaultDomains": "example.com",
+  "email_provider": "cloudflare",
+  "yyds_api_key": "",
+  "yyds_jwt": ""
+}

requirements.txt

@@ -0,0 +1,18 @@
+certifi==2026.4.22
+cffi==2.0.0
+charset-normalizer==3.4.7
+cssselect==1.3.0
+curl_cffi==0.13.0
+DataRecorder==3.6.2
+DownloadKit==2.0.7
+DrissionPage==4.1.1.2
+filelock==3.19.1
+idna==3.13
+lxml==6.1.0
+psutil==7.2.2
+pycparser==2.23
+requests==2.32.5
+requests-file==3.0.1
+tldextract==5.3.0
+urllib3==2.6.3
+websocket-client==1.9.0

sso_to_auth_json.py

@@ -0,0 +1,533 @@
+#!/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 热加载，可直接指向 CPA 挂载目录）",
+    )
+    ap.add_argument(
+        "--cpa-remote-url",
+        default=None,
+        help="远程 CPA 地址，如 http://IP: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())

tests/test_cloudflare_admin_api.py

@@ -0,0 +1,155 @@
+import unittest
+from unittest.mock import patch
+
+import cf_mail_debug
+import grok_register_ttk as app
+
+
+class DummyResponse:
+    def __init__(self, payload):
+        self._payload = payload
+        self.text = ""
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+class CloudflareAdminCreateTests(unittest.TestCase):
+    def setUp(self):
+        self.original_config = app.config.copy()
+        self.original_cf_domain_index = app._cf_domain_index
+        app._cf_domain_index = 0
+
+    def tearDown(self):
+        app.config = self.original_config
+        app._cf_domain_index = self.original_cf_domain_index
+
+    def test_default_config_keeps_cloudflare_temp_email_new_address(self):
+        app.config = app.DEFAULT_CONFIG.copy()
+        captured = {}
+
+        def fake_post(url, **kwargs):
+            captured["url"] = url
+            captured.update(kwargs)
+            return DummyResponse({"address": "anon@example.com", "jwt": "default-jwt"})
+
+        with patch.object(app, "http_post", side_effect=fake_post):
+            address, jwt = app.cloudflare_create_temp_address("https://temp-mail.example.com")
+
+        self.assertEqual(address, "anon@example.com")
+        self.assertEqual(jwt, "default-jwt")
+        self.assertEqual(captured["url"], "https://temp-mail.example.com/api/new_address")
+        self.assertEqual(captured["json"], {})
+        self.assertEqual(captured["headers"], {"Content-Type": "application/json"})
+
+    def test_app_uses_admin_new_address_with_x_admin_auth(self):
+        app.config.update({
+            "cloudflare_api_key": "admin-secret",
+            "cloudflare_auth_mode": "x-admin-auth",
+            "cloudflare_path_accounts": "/admin/new_address",
+            "defaultDomains": "vitassk.com",
+        })
+        captured = {}
+
+        def fake_post(url, **kwargs):
+            captured["url"] = url
+            captured.update(kwargs)
+            return DummyResponse({"address": "adminuser@vitassk.com", "jwt": "address-jwt"})
+
+        with patch.object(app, "generate_username", return_value="adminuser"), \
+                patch.object(app, "http_post", side_effect=fake_post):
+            address, jwt = app.cloudflare_create_temp_address("https://temp-mail.ikun.day")
+
+        self.assertEqual(address, "adminuser@vitassk.com")
+        self.assertEqual(jwt, "address-jwt")
+        self.assertEqual(captured["url"], "https://temp-mail.ikun.day/admin/new_address")
+        self.assertEqual(captured["json"], {
+            "name": "adminuser",
+            "domain": "vitassk.com",
+            "enablePrefix": True,
+        })
+        self.assertEqual(captured["headers"]["Content-Type"], "application/json")
+        self.assertEqual(captured["headers"]["x-admin-auth"], "admin-secret")
+
+    def test_app_keeps_anonymous_new_address_with_none_auth(self):
+        app.config.update({
+            "cloudflare_api_key": "",
+            "cloudflare_auth_mode": "none",
+            "cloudflare_custom_auth": "",
+            "cloudflare_path_accounts": "/api/new_address",
+            "defaultDomains": "vitassk.com",
+        })
+        captured = {}
+
+        def fake_post(url, **kwargs):
+            captured["url"] = url
+            captured.update(kwargs)
+            return DummyResponse({"address": "anon@vitassk.com", "jwt": "anon-jwt"})
+
+        with patch.object(app, "http_post", side_effect=fake_post):
+            address, jwt = app.cloudflare_create_temp_address("https://temp-mail.ikun.day")
+
+        self.assertEqual(address, "anon@vitassk.com")
+        self.assertEqual(jwt, "anon-jwt")
+        self.assertEqual(captured["url"], "https://temp-mail.ikun.day/api/new_address")
+        self.assertEqual(captured["json"], {"domain": "vitassk.com"})
+        self.assertEqual(captured["headers"], {"Content-Type": "application/json"})
+
+    def test_app_injects_custom_auth_on_anonymous_new_address(self):
+        app.config.update({
+            "cloudflare_api_key": "",
+            "cloudflare_auth_mode": "none",
+            "cloudflare_custom_auth": "global-pass",
+            "cloudflare_path_accounts": "/api/new_address",
+            "defaultDomains": "vitassk.com",
+        })
+        captured = {}
+
+        def fake_post(url, **kwargs):
+            captured["url"] = url
+            captured.update(kwargs)
+            return DummyResponse({"address": "anon@vitassk.com", "jwt": "anon-jwt"})
+
+        with patch.object(app, "http_post", side_effect=fake_post):
+            app.cloudflare_create_temp_address("https://temp-mail.ikun.day")
+
+        self.assertEqual(captured["headers"], {
+            "Content-Type": "application/json",
+            "x-custom-auth": "global-pass",
+        })
+
+    def test_debug_tool_can_create_address_through_admin_api(self):
+        captured = {}
+
+        def fake_post(url, **kwargs):
+            captured["url"] = url
+            captured.update(kwargs)
+            return DummyResponse({"address": "debuguser@vitassk.com", "jwt": "debug-jwt"})
+
+        with patch.object(cf_mail_debug.requests, "post", side_effect=fake_post):
+            address, jwt = cf_mail_debug.create_address(
+                "https://temp-mail.ikun.day",
+                auth_mode="x-admin-auth",
+                api_key="admin-secret",
+                create_path="/admin/new_address",
+                domain="vitassk.com",
+                name="debuguser",
+            )
+
+        self.assertEqual(address, "debuguser@vitassk.com")
+        self.assertEqual(jwt, "debug-jwt")
+        self.assertEqual(captured["url"], "https://temp-mail.ikun.day/admin/new_address")
+        self.assertEqual(captured["json"], {
+            "name": "debuguser",
+            "domain": "vitassk.com",
+            "enablePrefix": True,
+        })
+        self.assertEqual(captured["headers"]["Content-Type"], "application/json")
+        self.assertEqual(captured["headers"]["x-admin-auth"], "admin-secret")
+
+
+if __name__ == "__main__":
+    unittest.main()