Pine
32ca580a06
- sms stub 固定演示码123456 - 微信扫码(开放平台标准OAuth): /auth/wx-qr/start|wx-callback(GET/POST)|wx-qr/poll - 小程序扫码: /auth/mp-qr/start|poll|confirm - /auth/status 返回四 login_modes；WECHAT_OPEN_APPID/SECRET、AUTH_WECHAT_QR、WECHAT_QR_REDIRECT_URI 配置
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
# -*- 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
|