diff --git a/README.md b/README.md index 9802449..45ee032 100644 --- a/README.md +++ b/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。 ## 运行 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..aea6078 100755 --- a/config.example.json +++ b/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", diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 6e52bac..95c2998 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() @@ -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,24 +360,29 @@ 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"]} - resp_add = http_post( - f"{base}/tokens/add", - headers=headers, - params=query, - json=add_payload, - timeout=30, - proxies={}, - ) - resp_add.raise_for_status() - if log_callback: - log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({base}/tokens/add)") - return True - except Exception as add_exc: - if log_callback: - log_callback(f"[Debug] /tokens/add 写入失败，尝试 /tokens 全量模式: {add_exc}") + 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( + endpoint, + headers=headers, + params=query, + json=add_payload, + timeout=30, + proxies={}, + ) + resp_add.raise_for_status() + if log_callback: + 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 全量模式: {'; '.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() - 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..f99cd7e --- /dev/null +++ b/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() diff --git a/tests/test_grok2api_remote_pool.py b/tests/test_grok2api_remote_pool.py new file mode 100644 index 0000000..783752e --- /dev/null +++ b/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()