From 1bff08073812ce66f8c16cdfbb3fb424453373e1 Mon Sep 17 00:00:00 2001 From: VitaHuang Date: Sun, 5 Jul 2026 21:24:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(cloudflare):=20=E6=94=AF=E6=8C=81=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E5=91=98=E5=88=9B=E5=BB=BA=E4=B8=B4=E6=97=B6=E9=82=AE?= =?UTF-8?q?=E7=AE=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 28 +++++++++- cf_mail_debug.py | 79 ++++++++++++++++++++++++--- config.example.json | 8 +-- grok_register_ttk.py | 50 +++++++++++------ tests/__init__.py | 0 tests/test_cloudflare_admin_api.py | 87 ++++++++++++++++++++++++++++++ 6 files changed, 225 insertions(+), 27 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/test_cloudflare_admin_api.py diff --git a/README.md b/README.md index 9802449..8a0c9b1 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,9 @@ 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 文件路径 | @@ -94,6 +96,30 @@ cp config.example.json config.json | `grok2api_remote_base` | 远端 grok2api 管理 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。 ## 运行 diff --git a/cf_mail_debug.py b/cf_mail_debug.py index 3ce62e2..962960c 100755 --- a/cf_mail_debug.py +++ b/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", - json={}, - headers={"Content-Type": "application/json"}, + 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"/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: diff --git a/config.example.json b/config.example.json index a7ef20d..3b4413a 100755 --- a/config.example.json +++ b/config.example.json @@ -1,10 +1,10 @@ { "duckmail_api_key": "", - "cloudflare_api_base": "https://cloudflare.pages.dev", - "cloudflare_api_key": "", - "cloudflare_auth_mode": "none", + "cloudflare_api_base": "https://temp-mail.example.com", + "cloudflare_api_key": "your-admin-password", + "cloudflare_auth_mode": "x-admin-auth", "cloudflare_path_domains": "/api/domains", - "cloudflare_path_accounts": "/api/new_address", + "cloudflare_path_accounts": "/admin/new_address", "cloudflare_path_token": "/api/token", "cloudflare_path_messages": "/api/mails", "proxy": "", diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 6e52bac..f7bd9c0 100755 --- a/grok_register_ttk.py +++ b/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}""" - global _cf_domain_index - url = f"{api_base}/api/new_address" - payload = {} - try: - # 在多个域名之间轮换，降低单域偶发不收件导致的失败率 - domains = [x.strip() for x in str(config.get("defaultDomains", "") or "").split(",") if x.strip()] - if domains: - 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"}) + """适配 cloudflare_temp_email 新建地址接口并兼容 admin 创建模式。""" + path = get_cloudflare_path("cloudflare_path_accounts", "/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 = {} + if domain: + payload["domain"] = domain + headers = {"Content-Type": "application/json"} + resp = http_post(url, json=payload, headers=headers) resp.raise_for_status() try: data = resp.json() @@ -2480,7 +2501,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 +3010,3 @@ def main(): if __name__ == "__main__": main() - diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_cloudflare_admin_api.py b/tests/test_cloudflare_admin_api.py new file mode 100644 index 0000000..f2c69b5 --- /dev/null +++ b/tests/test_cloudflare_admin_api.py @@ -0,0 +1,87 @@ +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_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()