161 lines
6.2 KiB
Python
161 lines
6.2 KiB
Python
"""
|
|
邮箱验证码路由。
|
|
|
|
接口（统一响应格式 {success, data, message}）：
|
|
- POST /api/email/send-code 向邮箱发送验证码（10 分钟有效，60 秒内不可重复发送；
|
|
另有按邮箱/按 IP 的小时限流，防止被当作垃圾邮件中继）
|
|
- POST /api/email/verify-code 校验验证码（校验成功后即作废，一次性使用；
|
|
同一邮箱尝试超过 5 次自动作废验证码并要求重发）
|
|
|
|
注意：
|
|
- issue_code / verify_code 被密码重置路由复用，限流逻辑集中在两个函数内。
|
|
"""
|
|
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
from datetime import timedelta
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..email import send_verification_code
|
|
from ..models import PURPOSE_REGISTER, PURPOSE_RESET, EmailCode, utcnow
|
|
from ..schemas import SendCodeRequest, UnifiedResponse, VerifyCodeRequest
|
|
from ..security import (
|
|
get_client_ip,
|
|
is_valid_email,
|
|
send_code_email_limiter,
|
|
send_code_ip_limiter,
|
|
verify_code_limiter,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/email", tags=["email"])
|
|
|
|
# 验证码有效期与同一邮箱的重发间隔（可通过 .env 调整）
|
|
CODE_TTL_MINUTES = int(os.getenv("EMAIL_CODE_TTL_MINUTES") or "10")
|
|
RESEND_INTERVAL_SECONDS = int(os.getenv("EMAIL_RESEND_INTERVAL_SECONDS") or "60")
|
|
|
|
|
|
def issue_code(email: str, purpose: str, db: Session, request: Optional[Request] = None) -> None:
|
|
"""生成验证码并发送邮件（send-code 与忘记密码复用）；发送失败回滚并抛错。"""
|
|
now = utcnow()
|
|
|
|
# 限流：同一邮箱每小时最多 5 封；若带请求对象，再按 IP 每小时最多 10 封
|
|
if send_code_email_limiter.is_blocked(email):
|
|
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="该邮箱发送过于频繁，请稍后再试")
|
|
if request is not None:
|
|
client_ip = get_client_ip(request)
|
|
if send_code_ip_limiter.is_blocked(client_ip):
|
|
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="发送过于频繁，请稍后再试")
|
|
|
|
latest = (
|
|
db.query(EmailCode)
|
|
.filter(EmailCode.email == email, EmailCode.purpose == purpose)
|
|
.order_by(EmailCode.created_time.desc())
|
|
.first()
|
|
)
|
|
if latest is not None and latest.created_time > now - timedelta(seconds=RESEND_INTERVAL_SECONDS):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="发送过于频繁，请稍后再试")
|
|
|
|
code = f"{secrets.randbelow(1000000):06d}"
|
|
# 作废该邮箱同用途的历史验证码，只保留最新一条
|
|
db.query(EmailCode).filter(
|
|
EmailCode.email == email,
|
|
EmailCode.purpose == purpose,
|
|
).update({EmailCode.used: True})
|
|
|
|
record = EmailCode(
|
|
email=email,
|
|
code=code,
|
|
purpose=purpose,
|
|
expires_at=now + timedelta(minutes=CODE_TTL_MINUTES),
|
|
)
|
|
db.add(record)
|
|
db.flush()
|
|
try:
|
|
send_verification_code(email, code)
|
|
except Exception:
|
|
db.rollback()
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="验证码发送失败，请稍后再试")
|
|
db.commit()
|
|
|
|
# 发送成功：记录限流次数，并重置该校验尝试计数（新验证码重新计数）
|
|
send_code_email_limiter.hit(email)
|
|
if request is not None:
|
|
send_code_ip_limiter.hit(get_client_ip(request))
|
|
verify_code_limiter.reset(email)
|
|
|
|
|
|
def verify_code(payload: VerifyCodeRequest, db: Session) -> None:
|
|
"""校验验证码：正确则标记为已使用（一次性）；失败过多则作废验证码。"""
|
|
email = payload.email.strip().lower()
|
|
now = utcnow()
|
|
|
|
# 尝试限流：同一邮箱尝试超过 5 次即作废当前验证码，必须重新发送
|
|
if verify_code_limiter.is_blocked(email):
|
|
db.query(EmailCode).filter(
|
|
EmailCode.email == email,
|
|
EmailCode.purpose == payload.purpose,
|
|
EmailCode.used.is_(False),
|
|
).update({EmailCode.used: True})
|
|
db.commit()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail="验证码尝试次数过多，请重新发送",
|
|
)
|
|
|
|
record = (
|
|
db.query(EmailCode)
|
|
.filter(
|
|
EmailCode.email == email,
|
|
EmailCode.purpose == payload.purpose,
|
|
EmailCode.used.is_(False),
|
|
EmailCode.expires_at > now,
|
|
)
|
|
.order_by(EmailCode.created_time.desc())
|
|
.first()
|
|
)
|
|
# 恒定时间比较，避免通过响应时间差枚举验证码
|
|
if record is None or not hmac.compare_digest(record.code, payload.code):
|
|
verify_code_limiter.hit(email)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期")
|
|
|
|
record.used = True
|
|
verify_code_limiter.reset(email)
|
|
db.commit()
|
|
|
|
|
|
@router.post("/send-code", response_model=UnifiedResponse)
|
|
def send_code(
|
|
payload: SendCodeRequest,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
) -> UnifiedResponse:
|
|
"""向指定邮箱发送验证码（带邮箱与 IP 双重限流）。"""
|
|
email = payload.email.strip().lower()
|
|
if not is_valid_email(email):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="邮箱格式不正确")
|
|
if payload.purpose not in (PURPOSE_REGISTER, PURPOSE_RESET):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用途不合法")
|
|
|
|
issue_code(email, payload.purpose, db, request)
|
|
return UnifiedResponse(
|
|
success=True,
|
|
data={"email": email, "purpose": payload.purpose},
|
|
message="验证码已发送，请查收邮件",
|
|
)
|
|
|
|
|
|
@router.post("/verify-code", response_model=UnifiedResponse)
|
|
def verify_code_route(payload: VerifyCodeRequest, db: Session = Depends(get_db)) -> UnifiedResponse:
|
|
"""校验验证码：正确则标记为已使用（一次性）。"""
|
|
email = payload.email.strip().lower()
|
|
verify_code(payload, db)
|
|
return UnifiedResponse(
|
|
success=True,
|
|
data={"email": email, "purpose": payload.purpose},
|
|
message="验证码校验通过",
|
|
) |