Merge pull request #1 from kun775/cloudflare-admin-grok2api

feat(cloudflare): 支持管理员创建临时邮箱并兼容 grok2api 管理接口

README.md

@@ -86,14 +86,40 @@ cp config.example.json config.json
 | `proxy` | 代理地址，可留空 |
 | `enable_nsfw` | 注册后是否尝试开启 NSFW |
 | `cloudflare_api_base` | Cloudflare 临时邮箱 API 地址 |
-| `cloudflare_auth_mode` | Cloudflare API 鉴权模式：`none`、`bearer`、`x-api-key`、`query-key` |
+| `cloudflare_api_key` | Cloudflare 临时邮箱接口密钥；admin 模式填 `ADMIN_PASSWORD` |
+| `cloudflare_auth_mode` | Cloudflare API 鉴权模式：`none`、`bearer`、`x-api-key`、`x-admin-auth`、`query-key` |
+| `cloudflare_path_accounts` | Cloudflare 创建邮箱路径；匿名模式用 `/api/new_address`，admin 模式用 `/admin/new_address` |
 | `defaultDomains` | Cloudflare 临时邮箱默认域名 |
 | `grok2api_auto_add_local` | 是否写入本地 grok2api token 池 |
 | `grok2api_local_token_file` | 本地 grok2api token 文件路径 |
 | `grok2api_auto_add_remote` | 是否写入远端 grok2api |
-| `grok2api_remote_base` | 远端 grok2api 管理 API 地址 |
+| `grok2api_remote_base` | 远端 grok2api 地址，可填站点根地址或 `/admin/api` 管理 API 地址 |
 | `grok2api_remote_app_key` | 远端 grok2api app key |
 
+### Cloudflare 临时邮箱 admin 模式
+
+如果使用 `dreamhunter2333/cloudflare_temp_email` 且匿名 `/api/new_address` 开启了 Turnstile，可以改用 admin 创建邮箱接口：
+
+```json
+{
+  "email_provider": "cloudflare",
+  "cloudflare_api_base": "https://你的-worker-api-域名",
+  "cloudflare_api_key": "你的 ADMIN_PASSWORD",
+  "cloudflare_auth_mode": "x-admin-auth",
+  "cloudflare_path_accounts": "/admin/new_address",
+  "cloudflare_path_messages": "/api/mails",
+  "defaultDomains": "你的收信域名.com"
+}
+```
+
+创建邮箱会使用 `x-admin-auth` 调用 `/admin/new_address`，后续收件仍使用接口返回的地址 JWT 调用 `/api/mails`。
+
+可先用调试脚本验证 admin 创建接口：
+
+```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"
+```
+
 `config.json` 包含个人配置和密钥，不要提交到 Git。
 
 ## 运行

cf_mail_debug.py

@@ -3,6 +3,8 @@
 
 import argparse
 import re
+import secrets
+import string
 import time
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -36,21 +38,72 @@ def json_or_text(resp: requests.Response) -> Tuple[Optional[Dict[str, Any]], str
         return None, (resp.text or "")[:400]
 
 
-def create_address(api_base: str) -> Tuple[str, str]:
+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('/')}/api/new_address",
+        f"{api_base.rstrip('/')}{path}",
-        json={},
+        json=payload,
-        headers={"Content-Type": "application/json"},
+        headers=headers,
         timeout=20,
     )
     resp.raise_for_status()
     data, raw = json_or_text(resp)
     if not data:
-        raise RuntimeError(f"/api/new_address 非JSON: {raw}")
+        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"/api/new_address 缺少 address/jwt: {data}")
+        raise RuntimeError(f"{path} 缺少 address/jwt: {data}")
     return address, jwt
 
 
@@ -131,6 +184,11 @@ def main():
     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()
@@ -138,7 +196,14 @@ def main():
     address = args.address.strip()
     credential = args.credential.strip()
     if not credential:
-        address, credential = create_address(args.api_base)
+        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:

config.example.json

@@ -1,6 +1,6 @@
 {
   "duckmail_api_key": "",
-  "cloudflare_api_base": "https://cloudflare.pages.dev",
+  "cloudflare_api_base": "https://temp-mail.example.com",
   "cloudflare_api_key": "",
   "cloudflare_auth_mode": "none",
   "cloudflare_path_domains": "/api/domains",

grok_register_ttk.py

@@ -174,6 +174,8 @@ def cloudflare_build_headers(content_type=False):
     if key:
         if mode == "x-api-key":
             headers["X-API-Key"] = key
+        elif mode == "x-admin-auth":
+            headers["x-admin-auth"] = key
         elif mode != "none":
             headers["Authorization"] = f"Bearer {key}"
     return headers
@@ -188,6 +190,22 @@ def cloudflare_apply_auth_params(params=None):
     return merged
 
 
+def cloudflare_next_default_domain():
+    """按配置轮换选择 Cloudflare 临时邮箱域名。"""
+    global _cf_domain_index
+    domains = [x.strip() for x in str(config.get("defaultDomains", "") or "").split(",") if x.strip()]
+    if not domains:
+        return ""
+    domain = domains[_cf_domain_index % len(domains)]
+    _cf_domain_index += 1
+    return domain
+
+
+def cloudflare_is_admin_create_path(path):
+    """判断当前创建邮箱路径是否为 cloudflare_temp_email 管理员创建接口。"""
+    return str(path or "").rstrip("/").lower() == "/admin/new_address"
+
+
 def _pick_list_payload(data):
     if isinstance(data, list):
         return data
@@ -208,19 +226,22 @@ def _pick_list_payload(data):
 
 
 def cloudflare_create_temp_address(api_base):
-    """适配 cloudflare_temp_email v1.8.x: POST /api/new_address -> {address,jwt}"""
+    """适配 cloudflare_temp_email 新建地址接口并兼容 admin 创建模式。"""
-    global _cf_domain_index
+    path = get_cloudflare_path("cloudflare_path_accounts", "/api/new_address")
-    url = f"{api_base}/api/new_address"
+    url = f"{api_base}{path}"
+    domain = cloudflare_next_default_domain()
+    is_admin_create = cloudflare_is_admin_create_path(path)
+    if is_admin_create:
+        payload = {"name": generate_username(10), "enablePrefix": True}
+        if domain:
+            payload["domain"] = domain
+        headers = cloudflare_build_headers(content_type=True)
+    else:
         payload = {}
-    try:
+        if domain:
-        # 在多个域名之间轮换，降低单域偶发不收件导致的失败率
+            payload["domain"] = domain
-        domains = [x.strip() for x in str(config.get("defaultDomains", "") or "").split(",") if x.strip()]
+        headers = {"Content-Type": "application/json"}
-        if domains:
+    resp = http_post(url, json=payload, headers=headers)
-            payload["domain"] = domains[_cf_domain_index % len(domains)]
-            _cf_domain_index += 1
-    except Exception:
-        pass
-    resp = http_post(url, json=payload, headers={"Content-Type": "application/json"})
     resp.raise_for_status()
     try:
         data = resp.json()
@@ -295,6 +316,35 @@ def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None):
     return True
 
 
+def get_grok2api_remote_api_bases(base):
+    """生成 grok2api 管理 API 候选根路径。
+
+    参数:
+      - base str: 用户配置的 grok2api 远端地址
+
+    返回:
+      - list[str]: 依次尝试的管理 API 根路径
+    """
+    normalized = str(base or "").strip().rstrip("/")
+    if not normalized:
+        return []
+    lower = normalized.lower()
+    candidates = [normalized]
+    if lower.endswith("/admin/api"):
+        return candidates
+    if lower.endswith("/admin"):
+        candidates.append(f"{normalized}/api")
+    else:
+        candidates.append(f"{normalized}/admin/api")
+    seen = set()
+    unique = []
+    for item in candidates:
+        if item not in seen:
+            unique.append(item)
+            seen.add(item)
+    return unique
+
+
 def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None):
     token = _normalize_sso_token(raw_token)
     if not token:
@@ -310,11 +360,15 @@ def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None):
     query = {"app_key": app_key}
     pool_map = {"ssoBasic": "basic", "ssoSuper": "super"}
     remote_pool = pool_map.get(pool_name, "basic")
+    api_bases = get_grok2api_remote_api_bases(base)
+    add_errors = []
     # 优先使用 add 接口，避免全量覆盖远端池
-    try:
     add_payload = {"tokens": [token], "pool": remote_pool, "tags": ["auto-register"]}
+    for api_base in api_bases:
+        endpoint = f"{api_base}/tokens/add"
+        try:
             resp_add = http_post(
-            f"{base}/tokens/add",
+                endpoint,
                 headers=headers,
                 params=query,
                 json=add_payload,
@@ -323,11 +377,12 @@ def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None):
             )
             resp_add.raise_for_status()
             if log_callback:
-            log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({base}/tokens/add)")
+                log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({endpoint})")
             return True
         except Exception as add_exc:
+            add_errors.append(f"{endpoint}: {add_exc}")
     if log_callback:
-            log_callback(f"[Debug] /tokens/add 写入失败，尝试 /tokens 全量模式: {add_exc}")
+        log_callback(f"[Debug] /tokens/add 写入失败，尝试 /tokens 全量模式: {'; '.join(add_errors)}")
 
     # 兜底：旧版全量保存接口
     current = {}
@@ -2480,7 +2535,7 @@ class GrokRegisterGUI:
         add_label(2, 2, "Cloudflare 鉴权模式:")
         self.cloudflare_auth_mode_var = tk.StringVar(value=config.get("cloudflare_auth_mode", "bearer"))
         self.cloudflare_auth_mode_combo = tk_option_menu(
-            config_frame, self.cloudflare_auth_mode_var, ["query-key", "bearer", "x-api-key", "none"], width=12
+            config_frame, self.cloudflare_auth_mode_var, ["query-key", "bearer", "x-api-key", "x-admin-auth", "none"], width=12
         )
         add_field(self.cloudflare_auth_mode_combo, 2, 3, sticky=tk.W)
 
@@ -2989,4 +3044,3 @@ def main():
 
 if __name__ == "__main__":
     main()
-

tests/test_cloudflare_admin_api.py

@@ -0,0 +1,110 @@
+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()
+
+    def tearDown(self):
+        app.config = self.original_config
+
+    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_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_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()

tests/test_grok2api_remote_pool.py

@@ -0,0 +1,81 @@
+import unittest
+from unittest.mock import patch
+
+import grok_register_ttk as app
+
+
+class DummyResponse:
+    def __init__(self, payload=None, status_code=200, reason=""):
+        self._payload = payload or {}
+        self.status_code = status_code
+        self.reason = reason
+        self.text = ""
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP Error {self.status_code}: {self.reason}")
+
+    def json(self):
+        return self._payload
+
+
+class Grok2ApiRemotePoolTests(unittest.TestCase):
+    def setUp(self):
+        self.original_config = app.config.copy()
+
+    def tearDown(self):
+        app.config = self.original_config
+
+    def test_remote_pool_falls_back_to_admin_api_prefix_when_root_tokens_add_is_404(self):
+        app.config.update({
+            "grok2api_remote_base": "https://grok.example.com",
+            "grok2api_remote_app_key": "app-secret",
+            "grok2api_pool_name": "ssoBasic",
+        })
+        calls = []
+
+        def fake_post(url, **kwargs):
+            calls.append((url, kwargs))
+            if url == "https://grok.example.com/tokens/add":
+                return DummyResponse(status_code=404)
+            return DummyResponse({"status": "success", "count": 1})
+
+        with patch.object(app, "http_post", side_effect=fake_post):
+            ok = app.add_token_to_grok2api_remote_pool("sso=abc123", email="a@example.com")
+
+        self.assertTrue(ok)
+        self.assertEqual([url for url, _ in calls], [
+            "https://grok.example.com/tokens/add",
+            "https://grok.example.com/admin/api/tokens/add",
+        ])
+        self.assertEqual(calls[-1][1]["params"], {"app_key": "app-secret"})
+        self.assertEqual(calls[-1][1]["json"], {
+            "tokens": ["abc123"],
+            "pool": "basic",
+            "tags": ["auto-register"],
+        })
+
+    def test_remote_pool_does_not_duplicate_admin_api_prefix_when_base_already_points_to_admin_api(self):
+        app.config.update({
+            "grok2api_remote_base": "https://grok.example.com/admin/api",
+            "grok2api_remote_app_key": "app-secret",
+            "grok2api_pool_name": "ssoSuper",
+        })
+        calls = []
+
+        def fake_post(url, **kwargs):
+            calls.append((url, kwargs))
+            return DummyResponse({"status": "success", "count": 1})
+
+        with patch.object(app, "http_post", side_effect=fake_post):
+            ok = app.add_token_to_grok2api_remote_pool("sso=super123", email="a@example.com")
+
+        self.assertTrue(ok)
+        self.assertEqual([url for url, _ in calls], [
+            "https://grok.example.com/admin/api/tokens/add",
+        ])
+        self.assertEqual(calls[0][1]["json"]["pool"], "super")
+
+
+if __name__ == "__main__":
+    unittest.main()