shaguabijia-app-server/app/integrations/sms/chuanglan.py

"""创蓝云智(253)短信 provider(自管码 Mode B)。

创蓝 `tpl/send` v2 是**纯发送网关**(本服务生成码 → 放入 templateParamJson → 创蓝只下发,
无校验接口),故与极光同为 **Mode B**:本服务生成/存储/校验验证码,创蓝只负责发。

**本模块的存码/冷却/一次性/防爆破/GC 机器与 [jiguang.py](jiguang.py) 是刻意的隔离复制**
(设计见 docs/superpowers/specs/2026-07-26-chuanglan-sms-verify-design.md):极光文件一行不动、
零回归风险于登录关键路径的默认 provider;代价是两处 Mode B 并发逻辑重复,改动需同步。唯一新逻辑
是 `_send_via_chuanglan`(HMAC-SHA256 签名 + httpx POST + 错误码映射)。

两种运行模式由 `SMS_MOCK` 切换:
  - **mock**(开发/测试,默认):不真发,验证码打日志;校验放行任意 N 位数字。
  - **real**(`SMS_MOCK=false` 且 `SMS_PROVIDER=chuanglan`):`secrets` 生成码 → 调创蓝 `tpl/send`
    下发(HMAC 签名,password 仅本地算签不上行)→ 校验比对本地存码(一次性 / 过期 / 防爆破)。

验证码存储:**进程内存**(单 worker 够用,多 worker 不共享,与极光同级技术债)。防刷同极光:
单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)+ 单设备/IP 频控(api 层)+ 单码失败 `SMS_MAX_VERIFY_ATTEMPTS`
次即作废。运维侧另需在创蓝控制台配 **IP 白名单**(否则 117)。接口调研见 docs/integrations/chuanglan/tpl-send.md。
"""
from __future__ import annotations

import hashlib
import hmac
import json
import logging
import secrets
import time
from dataclasses import dataclass
from threading import Lock

import httpx

from app.core.config import settings

from .base import SmsError, mock_verify

logger = logging.getLogger("shagua.sms.chuanglan")


@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 时顺手清过期项(防无限增长)

# 发码错误码(创蓝 `code`)→ (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
    "103": (429, "发送过于频繁，请稍后再试"),   # 提交速度过快
    "107": (400, "手机号无效"),                 # 手机号码错误
}
# 需运维介入的配置/开通/余额类错误:打 critical 日志(仍归 503)。
_SEND_CRITICAL_CODES = frozenset({
    "109",  # 无发送量/余额不足
    "117",  # IP 未加白名单
    "102",  # 密码错误
    "116",  # 签名不合法
    "124",  # 模板内容不匹配
    "152",  # 模板不存在
    "101",  # 账号不存在
    "118",  # 无发送权限
})


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 超阈值才扫它。"""
    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 / 手机号无效 400 / 供应商失败 503)
    """
    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-chuanglan-MOCK] to %s**** code=%s (不真发)", phone[:3], code)
        else:
            _send_via_chuanglan(phone, code)
            logger.info("[SMS-chuanglan] sent to %s****", phone[:3])
    except Exception as e:
        # 发送失败:**保留冷却**(失败也限速,挡住余额不足/签名失效时前端重试狂打),
        # 只清掉没发出去的码(用户收不到,留着无意义且占内存)。
        with _lock:
            _codes.pop(phone, None)
        if isinstance(e, SmsError):
            raise
        logger.exception("[SMS-chuanglan] 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 = mock_verify(code)
        logger.info("[SMS-chuanglan-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


# ============================ 发送接缝(单测 monkeypatch 这两个 / httpx.post)============================

def _sign(password: str, timestamp: str, nonce: str) -> str:
    """创蓝 HMAC-SHA256 签名:key=md5(password),msg=sorted([md5pwd,ts,nonce]) 拼接去空白,输出小写 hex。"""
    md5pwd = hashlib.md5(password.encode()).hexdigest()          # 32 位小写 hex
    raw = "".join(sorted([md5pwd, timestamp, nonce]))            # 字典序升序,无分隔符拼接
    raw = "".join(raw.split())                                  # 去所有空白(faithful;三段本无空白)
    return hmac.new(md5pwd.encode(), raw.encode(), hashlib.sha256).hexdigest()


def _call_chuanglan(phone: str, code: str) -> dict:
    """组装 + 签名 + POST 创蓝 tpl/send,返回解析后的响应 dict。

    传输错误 / HTTP≠200 / 响应非 JSON 一律抛 SmsError(503)(保「provider 出问题→503」不变式);
    业务码(含 000000)由调用方 `_send_via_chuanglan` 判读。password 只用于算签,不入 body。
    """
    timestamp = str(int(time.time()))
    nonce = secrets.token_hex(16)  # 32 位 hex
    body = {
        "account": settings.CHUANGLAN_SMS_ACCOUNT,
        "timestamp": timestamp,
        "nonce": nonce,
        "phoneNumbers": phone,
        "templateId": settings.CHUANGLAN_SMS_TEMPLATE_ID,
        "templateParamJson": json.dumps([{"param1": code}]),
    }
    if settings.CHUANGLAN_SMS_SIGNATURE:
        body["signature"] = settings.CHUANGLAN_SMS_SIGNATURE
    headers = {
        "Content-Type": "application/json",
        "X-QA-Hmac-Signature": _sign(settings.CHUANGLAN_SMS_PASSWORD, timestamp, nonce),
    }
    try:
        resp = httpx.post(
            settings.CHUANGLAN_SMS_ENDPOINT,
            json=body,
            headers=headers,
            timeout=settings.CHUANGLAN_SMS_TIMEOUT_SEC,
        )
    except httpx.HTTPError as e:
        logger.exception("[SMS-chuanglan] 网络错误 phone=%s****", phone[:3])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503) from e

    if resp.status_code != 200:
        logger.error("[SMS-chuanglan] http=%s body=%s", resp.status_code, resp.text[:200])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503)
    try:
        return resp.json()
    except Exception as e:
        logger.error("[SMS-chuanglan] 响应非 JSON: %s", resp.text[:200])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503) from e


def _send_via_chuanglan(phone: str, code: str) -> None:
    """调创蓝 tpl/send 发送。成功静默返回;失败按错误码映射抛 SmsError。"""
    if not settings.chuanglan_sms_configured:
        raise SmsError("短信服务未配置(缺创蓝 account/password/templateId)", status_code=503)

    result = _call_chuanglan(phone, code)  # 传输/非200/解析异常在内部抛 SmsError(503)
    rcode = str(result.get("code"))
    if rcode == "000000":
        return

    emsg = result.get("errorMsg") or ""
    logger.error("[SMS-chuanglan] send failed code=%s msg=%s", rcode, emsg)
    if rcode in _SEND_CRITICAL_CODES:
        logger.critical("[SMS-chuanglan] %s —— 需运维处理(余额/IP白名单/密码/签名/模板/账号)", rcode)
    status, msg = _SEND_ERRORS.get(rcode, (503, "短信服务暂不可用,请稍后重试"))
    raise SmsError(msg, status_code=status)