# -*- coding: utf-8 -*- """短信验证码服务(统一登录)。 默认 ``stub`` provider 仅打印验证码不发真实短信,便于无短信资质联调; 接入真实短信(阿里云 / 腾讯云)时新增 provider 分支即可,接口不变。 验证码存内存(带 TTL 与尝试次数),进程重启即失效——生产应换 Redis 存储(见 云超服一体化架构设计方案 §8.4 缓存服务)。 """ from __future__ import annotations import secrets import threading import time from .. import config # phone -> {"code": str, "expires_at": float, "attempts": int} _STORE: dict[str, dict] = {} _LOCK = threading.Lock() _CODE_TTL = config.SMS_CODE_TTL_SECONDS _MAX_ATTEMPTS = config.SMS_RATE_LIMIT class SmsError(Exception): """短信验证码业务异常(校验失败 / 限流等)。""" def _send(phone: str, code: str) -> None: """按 provider 发送验证码。stub 只打印,供联调与测试。""" if config.SMS_PROVIDER == "stub": print(f"[sms:stub] verification-code -> {phone}: {code}") return # 预留:接入阿里云 / 腾讯云短信 SDK(环境注入凭据),按 provider 分发。 raise SmsError(f"unsupported sms provider: {config.SMS_PROVIDER}") # 演示验证码:stub provider 统一用固定码(先跑通流程,生产换真实短信)。 DEMO_CODE = "123456" def issue(phone: str) -> str: """为手机号生成并发送 6 位验证码,返回明文(stub 下发时打印)。 演示期 stub provider 统一返回固定码 ``123456``(见配置 SMS_PROVIDER=stub), 便于四端联调;接入真实短信后按 provider 生成随机码。 """ code = DEMO_CODE if config.SMS_PROVIDER == "stub" else f"{secrets.randbelow(1_000_000):06d}" with _LOCK: _STORE[phone] = { "code": code, "expires_at": time.monotonic() + _CODE_TTL, "attempts": 0, } _send(phone, code) return code def verify(phone: str, code: str) -> bool: """校验验证码;成功即作废,失败累计次数超限则不可再试。""" with _LOCK: rec = _STORE.get(phone) if not rec: raise SmsError("verification code not issued") if time.monotonic() > rec["expires_at"]: _STORE.pop(phone, None) raise SmsError("verification code expired") if rec["attempts"] >= _MAX_ATTEMPTS: raise SmsError("too many attempts, request a new code") if rec["code"] != code.strip(): rec["attempts"] += 1 raise SmsError("invalid verification code") _STORE.pop(phone, None) return True