fix(register): harden OTP stage gating and adult profile generation

Author: qwe12369

src/config/constants.py

@@ -3,7 +3,7 @@
 """
 
 import random
-from datetime import datetime
+from datetime import datetime, date, timedelta
 from enum import Enum
 from typing import Dict, List, Tuple
 
@@ -282,20 +282,26 @@ def generate_random_user_info() -> dict:
     # 注册阶段使用更像真人资料的姓名格式，降低单名触发风控的概率。
     name = f"{random.choice(FIRST_NAMES)} {random.choice(LAST_NAMES)}"
 
-    # 生成随机生日（18-45岁）
-    current_year = datetime.now().year
-    birth_year = random.randint(current_year - 45, current_year - 18)
-    birth_month = random.randint(1, 12)
-    # 根据月份确定天数
-    if birth_month in [1, 3, 5, 7, 8, 10, 12]:
-        birth_day = random.randint(1, 31)
-    elif birth_month in [4, 6, 9, 11]:
-        birth_day = random.randint(1, 30)
-    else:
-        # 2月，简化处理
-        birth_day = random.randint(1, 28)
-
-    birthdate = f"{birth_year}-{birth_month:02d}-{birth_day:02d}"
+    # 生成随机生日（按“当天年龄”精确计算，避免出现未满 18 周岁的边界值）。
+    # 这里默认 21~45 岁，降低生日边界导致的注册风险。
+    min_age = 21
+    max_age = 45
+
+    today = date.today()
+
+    def _years_ago(base: date, years: int) -> date:
+        try:
+            return base.replace(year=base.year - years)
+        except ValueError:
+            # 兼容 2/29
+            return base.replace(month=2, day=28, year=base.year - years)
+
+    latest_birth = _years_ago(today, min_age)   # 最年轻（也已满 min_age）
+    oldest_birth = _years_ago(today, max_age)   # 最年长
+    span_days = max((latest_birth - oldest_birth).days, 0)
+    random_offset = random.randint(0, span_days)
+    birth_day = oldest_birth + timedelta(days=random_offset)
+    birthdate = birth_day.strftime("%Y-%m-%d")
 
     return {
         "name": name,

src/core/register.py

@@ -1741,8 +1741,10 @@ def _complete_chatgpt_bridge_stage(self, result: RegistrationResult, *, final_ur
                 "warning" if is_chatgpt_stage else "info",
             )
             if is_chatgpt_stage:
-                self._log(f"{log_prefix}先主动重发一次 OTP，确保当前 bridge 阶段拿到新验证码...", "warning")
-                self._send_verification_code(referer="https://auth.openai.com/email-verification")
+                self._log(
+                    f"{log_prefix}优先等待当前阶段系统自动发送的 OTP，避免过早重发导致旧验证码混入...",
+                    "warning",
+                )
                 self._prepare_expected_login_otp(grace_seconds=OTP_CREATED_AT_GRACE_SECONDS)
         elif "auth.openai.com/log-in/password" in current_url:
             self._log(f"{log_prefix}已直达密码页，先提交密码再继续...")
@@ -2641,12 +2643,6 @@ def _complete_token_exchange_native_backup(self, result: RegistrationResult) ->
         原生入口对齐备份版收尾链路：
         登录验证码 -> Workspace -> redirect -> OAuth callback -> token 入袋。
         """
-        def _is_registration_gate_url(url: str) -> bool:
-            u = str(url or "").strip().lower()
-            if not u:
-                return False
-            return ("auth.openai.com/about-you" in u) or ("auth.openai.com/add-phone" in u)
-
         if not self._complete_login_otp_once(result, fetch_timeout=120):
             return False
 
@@ -2667,12 +2663,12 @@ def _is_registration_gate_url(url: str) -> bool:
             result.workspace_id = workspace_id
 
         continue_url = ""
-        if otp_continue and _is_registration_gate_url(otp_continue):
+        if otp_continue and self._is_registration_gate_url(otp_continue):
             self._log("OTP 返回 continue_url 指向注册门页（about-you/add-phone），本轮收尾忽略该地址")
             otp_continue = ""
 
         cached_continue = str(self._create_account_continue_url or "").strip()
-        if cached_continue and _is_registration_gate_url(cached_continue):
+        if cached_continue and self._is_registration_gate_url(cached_continue):
             self._log("create_account 缓存 continue_url 指向注册门页（about-you/add-phone），本轮收尾忽略该地址")
             cached_continue = ""
 
@@ -3669,6 +3665,35 @@ def _verify_email_otp_with_retry(
                 attempted_codes.add(code)
 
             if self._validate_verification_code(code):
+                if chatgpt_stage:
+                    continue_url = str(self._last_validate_otp_continue_url or "").strip()
+                    lowered_continue = continue_url.lower()
+
+                    # ChatGPT 补会话阶段命中注册门页通常是旧会话链路，不应视为成功。
+                    if self._is_registration_gate_url(continue_url):
+                        if (not resend_after_invalid_triggered) and attempt < max_attempts:
+                            resend_after_invalid_triggered = True
+                            self._log(
+                                f"{stage_label}命中注册门页 continue_url（about-you/add-phone），疑似旧会话验证码；主动重发一次 OTP...",
+                                "warning",
+                            )
+                            self._send_verification_code(referer="https://auth.openai.com/email-verification")
+                            self._prepare_expected_login_otp(grace_seconds=OTP_CREATED_AT_GRACE_SECONDS)
+
+                        if attempt < max_attempts:
+                            self._log(
+                                f"{stage_label}校验成功但 continue_url 仍指向注册门页，继续等待当前登录阶段新验证码...",
+                                "warning",
+                            )
+                            time.sleep(2)
+                            continue
+                        return False
+
+                    # 回调明确 access_denied 时直接结束当前阶段，避免无意义循环。
+                    if ("chatgpt.com/api/auth/callback/openai" in lowered_continue) and ("error=access_denied" in lowered_continue):
+                        self._log(f"{stage_label}回调返回 access_denied，终止当前验证码阶段重试", "warning")
+                        return False
+
                 return True
 
             if chatgpt_stage:
@@ -3909,16 +3934,19 @@ def _resolve_workspace_selection_referer(self, referer_url: Optional[str] = None
             or "https://auth.openai.com/sign-in-with-chatgpt/codex/consent"
         ).strip()
 
+    @staticmethod
+    def _is_registration_gate_url(url: str) -> bool:
+        lowered = str(url or "").strip().lower()
+        if not lowered:
+            return False
+        return ("auth.openai.com/about-you" in lowered) or ("auth.openai.com/add-phone" in lowered)
+
     def _fetch_workspace_selection_context(
         self,
         referer_url: Optional[str] = None,
         *,
         force_probe: bool = False,
     ) -> Tuple[str, str, Dict[str, Any]]:
-        def _is_registration_gate_url(url: str) -> bool:
-            lowered = str(url or "").strip().lower()
-            return bool(lowered) and ("about-you" in lowered or "add-phone" in lowered)
-
         raw_referer = str(
             referer_url
             or self._last_validate_otp_continue_url
@@ -3927,7 +3955,7 @@ def _is_registration_gate_url(url: str) -> bool:
         ).strip()
         consent_page_url = raw_referer or self._resolve_workspace_selection_referer(referer_url)
         has_explicit_consent_url = _looks_like_codex_consent_url(raw_referer)
-        should_probe_default_consent = _is_registration_gate_url(consent_page_url) or (
+        should_probe_default_consent = self._is_registration_gate_url(consent_page_url) or (
             force_probe and not has_explicit_consent_url
         )
         if (not consent_page_url) or should_probe_default_consent:
@@ -3945,7 +3973,7 @@ def _is_registration_gate_url(url: str) -> bool:
                         timeout=20,
                     )
                     response_url = str(getattr(response, "url", request_url) or request_url).strip()
-                    if response_url and ("localhost:1455" not in response_url.lower()) and (not _is_registration_gate_url(response_url)):
+                    if response_url and ("localhost:1455" not in response_url.lower()) and (not self._is_registration_gate_url(response_url)):
                         consent_page_url = response_url
                     consent_text = str(getattr(response, "text", "") or "")
                 except Exception as exc:

src/services/luckmail_mail.py

@@ -124,6 +124,10 @@ def __init__(self, config: Dict[str, Any] = None, name: str = None):
         self._orders_by_email: Dict[str, Dict[str, Any]] = {}
         # 记录每个订单/Token 最近返回过的验证码，避免后续阶段反复拿到旧码。
         self._recent_codes_by_order: Dict[str, Dict[str, float]] = {}
+        # 记录每个订单/Token 最近一次成功使用的 message_id，避免重复消费旧邮件。
+        self._last_used_message_ids: Dict[str, str] = {}
+        # 最近一次成功取码的元信息，供上层做验证码去重。
+        self.last_verification_meta: Dict[str, Any] = {}
         self._data_dir = Path(__file__).resolve().parents[2] / "data"
         self._registered_file = self._data_dir / "luckmail_registered_emails.json"
         self._failed_file = self._data_dir / "luckmail_failed_emails.json"
@@ -198,6 +202,53 @@ def _remember_code(self, order_key: str, code: str, now: Optional[float] = None)
             for key in stale:
                 order_cache.pop(key, None)
 
+    @staticmethod
+    def _parse_mail_timestamp(value: Any) -> Optional[float]:
+        text = str(value or "").strip()
+        if not text:
+            return None
+        # epoch / epoch_ms
+        try:
+            numeric = float(text)
+            if numeric > 10_000_000_000:  # ms
+                numeric = numeric / 1000.0
+            if numeric > 0:
+                return numeric
+        except Exception:
+            pass
+
+        # ISO datetime
+        normalized = text.replace("Z", "+00:00")
+        try:
+            dt = datetime.fromisoformat(normalized)
+            if dt.tzinfo is None:
+                dt = dt.replace(tzinfo=timezone.utc)
+            return dt.timestamp()
+        except Exception:
+            return None
+
+    def _extract_token_mail_meta(self, result: Any) -> Dict[str, Any]:
+        mail_payload = self._extract_field(result, "mail") or {}
+        if not isinstance(mail_payload, dict):
+            mail_payload = {}
+        message_id = str(
+            mail_payload.get("message_id")
+            or mail_payload.get("id")
+            or mail_payload.get("mail_id")
+            or ""
+        ).strip()
+        received_at_raw = (
+            mail_payload.get("received_at")
+            or mail_payload.get("created_at")
+            or mail_payload.get("timestamp")
+            or ""
+        )
+        mail_ts = self._parse_mail_timestamp(received_at_raw)
+        return {
+            "message_id": message_id,
+            "mail_ts": mail_ts,
+        }
+
     def _now_iso(self) -> str:
         return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
@@ -904,6 +955,7 @@ def get_verification_code(
         pattern: str = OTP_CODE_PATTERN,
         otp_sent_at: Optional[float] = None,
     ) -> Optional[str]:
+        self.last_verification_meta = {}
         order_info = self._find_order(email=email, email_id=email_id)
 
         token = ""
@@ -970,6 +1022,21 @@ def get_verification_code(
                     return None
 
                 now_ts = time.time()
+                mail_meta = self._extract_token_mail_meta(result)
+                message_id = str(mail_meta.get("message_id") or "").strip()
+                mail_ts = mail_meta.get("mail_ts")
+
+                if otp_sent_at and (mail_ts is not None) and (mail_ts + 2 < float(otp_sent_at)):
+                    # 明确早于本轮发码时间窗口，判定旧邮件。
+                    time.sleep(poll_interval)
+                    continue
+
+                if message_id:
+                    last_message_id = str(self._last_used_message_ids.get(code_key) or "").strip()
+                    if last_message_id and message_id == last_message_id:
+                        time.sleep(poll_interval)
+                        continue
+
                 if otp_guard_until and now_ts < otp_guard_until and self._is_recent_code(code_key, code, now_ts):
                     time.sleep(poll_interval)
                     continue
@@ -980,11 +1047,20 @@ def get_verification_code(
                     continue
 
                 self._remember_code(code_key, code, now_ts)
+                if message_id:
+                    self._last_used_message_ids[code_key] = message_id
+                self.last_verification_meta = {
+                    "mail_id": message_id or code_key,
+                    "mail_ts": mail_ts if mail_ts is not None else now_ts,
+                    "code": code,
+                    "fingerprint": f"{mail_ts if mail_ts is not None else now_ts}|{message_id or code_key}|{code}",
+                }
                 self.update_status(True)
                 return code
 
             time.sleep(poll_interval)
 
+        self.last_verification_meta = {}
         return None
 
     def list_emails(self, **kwargs) -> List[Dict[str, Any]]:

src/web/routes/email.py

@@ -325,18 +325,6 @@ async def get_service_types():
                     {"name": "domain", "label": "邮箱域名", "required": False, "placeholder": "example.com"},
                 ]
             },
-            {
-                "value": "cloudmail",
-                "label": "CloudMail",
-                "description": "CloudMail 自部署 Cloudflare Worker 邮箱服务，使用管理口令创建邮箱并轮询验证码",
-                "config_fields": [
-                    {"name": "base_url", "label": "API 地址", "required": True, "placeholder": "https://cloudmail.example.com"},
-                    {"name": "admin_password", "label": "Admin 密码", "required": True, "secret": True},
-                    {"name": "domain", "label": "邮箱域名", "required": True, "placeholder": "example.com"},
-                    {"name": "enable_prefix", "label": "启用前缀", "required": False, "default": True},
-                    {"name": "timeout", "label": "超时时间", "required": False, "default": 30},
-                ]
-            },
             {
                 "value": "imap_mail",
                 "label": "IMAP 邮箱",