#!/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()