"""短信验证码服务。 两种运行模式由 `SMS_MOCK` 切换: - **mock**(开发/测试,默认):不真发短信,验证码打到日志;校验**放行任意 N 位数字** (测试/开发便利)。真实校验逻辑(比对存码 / 一次性 / 防爆破)由 real 分支 + 单测覆盖。 - **real**(生产 `SMS_MOCK=false`):本服务生成 N 位验证码 → 调极光短信 REST `/v1/messages` 发送(自定义验证码模式,极光只负责发,code 由本服务生成/保管/ 校验)→ 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)。 验证码存储:**进程内存**(单 worker uvicorn 够用)。重启丢失(用户重发即可)。多 worker / 多机时内存不共享 → 冷却、校验都会失效,届时迁移到 DB/Redis。 见 docs/待办与技术债.md。 防刷两层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权): 1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件) 2. 单设备(device_id)每小时频控(api 层 auth.sms_send 内 enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)。 ⚠️ 原「单 IP 频控(rate_limit 依赖)」2026-06-26 按产品要求删除、改设备维度;但 device_id 客户端可伪造/轮换, 脚本轮换 id 能绕过本层 → 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)。 ⚠️ 原「单号每日上限」2026-07-03 按精简要求删除(mentor 定:登录风控只留单号冷却 + 单设备频控); 单号维度现仅剩 60s 冷却,「换号轰炸」由单设备频控封顶。 另:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)。 """ from __future__ import annotations import base64 import logging import secrets import time from dataclasses import dataclass from threading import Lock import httpx from app.core.config import settings logger = logging.getLogger("shagua.sms") class SmsError(Exception): """业务异常。`status_code` 决定 api 层翻成哪个 HTTP 码: 过频/每日超限 = 429(客户端等会再来),供应商不可用 = 503,手机号无效 = 400。 """ def __init__(self, message: str, status_code: int = 429) -> None: super().__init__(message) self.status_code = status_code @dataclass class _CodeRecord: code: str expires_at: float attempts: int = 0 # 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring) _codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码 _last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却) _lock = Lock() _GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit) def _gen_code() -> str: """生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。""" return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH)) def _gc(now: float) -> None: """顺手清理过期内存项,防两个 dict 无限增长。仅在持锁时调用,且某 dict 超 _GC_THRESHOLD 才扫它(低频,开销可忽略)。""" if len(_codes) > _GC_THRESHOLD: for p in [p for p, r in _codes.items() if now > r.expires_at]: _codes.pop(p, None) if len(_last_sent) > _GC_THRESHOLD: cutoff = now - settings.SMS_SEND_INTERVAL_SEC for p in [p for p, ts in _last_sent.items() if ts < cutoff]: _last_sent.pop(p, None) def send_code(phone: str) -> int: """发送验证码。 Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC) Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400) """ now = time.time() # --- lock 内:防刷检查 + 预占(防并发重复发烧钱)--- with _lock: _gc(now) # 顺手清过期内存(超阈值才扫) elapsed = now - _last_sent.get(phone, 0.0) if elapsed < settings.SMS_SEND_INTERVAL_SEC: remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed) raise SmsError(f"发送过于频繁,请 {remain}s 后再试") code = _gen_code() # 预占:先记冷却/存码,释放锁后再发网络(发失败保留冷却,见下) _last_sent[phone] = now _codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC) # --- lock 外:真正发送(网络 IO 不持锁)--- try: if settings.SMS_MOCK: logger.info("[SMS-MOCK] to %s**** code=%s (不真发)", phone[:3], code) else: _send_via_jiguang(phone, code) logger.info("[SMS] sent to %s****", phone[:3]) except Exception as e: # 发送失败:**保留冷却**(失败也限速,挡住余额不足/签名失效时 # 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。 with _lock: _codes.pop(phone, None) if isinstance(e, SmsError): raise logger.exception("[SMS] send failed phone=%s****", phone[:3]) raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e return settings.SMS_SEND_INTERVAL_SEC def verify_code(phone: str, code: str) -> bool: """校验验证码。 - **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。 - **real 模式**:比对本服务存的码,匹配即作废(一次性);失败累计到上限也作废(防爆破)。 """ if settings.SMS_MOCK: ok = len(code) == settings.SMS_CODE_LENGTH and code.isdigit() logger.info("[SMS-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3]) return ok with _lock: rec = _codes.get(phone) if rec is None: return False if time.time() > rec.expires_at: _codes.pop(phone, None) return False if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS: _codes.pop(phone, None) # 试错过多,作废 return False if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")): _codes.pop(phone, None) # 验过即作废 return True rec.attempts += 1 return False def _send_via_jiguang(phone: str, code: str) -> None: """调极光短信 REST /v1/messages 发送(自定义验证码模式)。失败抛 SmsError。""" if not settings.JG_APP_KEY or not settings.JG_MASTER_SECRET: raise SmsError("短信服务未配置(缺 JG_APP_KEY/JG_MASTER_SECRET)", status_code=503) if not settings.SMS_SIGN_ID or not settings.SMS_TEMPLATE_ID: raise SmsError("短信服务未配置(缺 SMS_SIGN_ID/SMS_TEMPLATE_ID)", status_code=503) auth_b64 = base64.b64encode( f"{settings.JG_APP_KEY}:{settings.JG_MASTER_SECRET}".encode() ).decode() body = { "mobile": phone, "sign_id": settings.SMS_SIGN_ID, "temp_id": settings.SMS_TEMPLATE_ID, "temp_para": {"code": code}, } try: resp = httpx.post( settings.SMS_SEND_ENDPOINT, json=body, headers={ "Content-Type": "application/json", "Authorization": f"Basic {auth_b64}", }, timeout=settings.JG_REQUEST_TIMEOUT_SEC, ) except httpx.HTTPError as e: raise SmsError(f"短信网关网络错误: {e}", status_code=503) from e if resp.status_code == 200: return # {"msg_id": ...} # 极光错误码映射(节选常见;完整码表见 docs/integrations/sms.md) try: err = resp.json().get("error", {}) ecode, emsg = err.get("code"), err.get("message", "") except Exception: ecode, emsg = None, resp.text[:200] logger.error("[SMS] jiguang error http=%s code=%s msg=%s", resp.status_code, ecode, emsg) if ecode == 50014: # 余额不足:运维要立即告警充值 logger.critical("[SMS] 极光短信余额不足(50014),需充值!") raise SmsError("短信服务暂不可用,请稍后重试", status_code=503) if ecode == 50009: # 极光侧超频 raise SmsError("发送过于频繁,请稍后再试", status_code=429) if ecode == 50006: # 手机号无效(schema 已挡格式,这里多是空号/停机) raise SmsError("手机号无效", status_code=400) raise SmsError(f"短信发送失败(code={ecode})", status_code=503)