feat(BIZ-129): Flask-Limiter 集成 + 登录/重置限流

- requirements.txt: flask-limiter>=4.0
- app.py: 创建 Limiter 实例(memory:// 存储,60/分钟默认)
- auth_api.py:
  - 简化 register_auth_api(app, limiter=None)——limiter 仅作能力标记
  - _rate_limit(rate, methods) 装饰器:手写 IP+endpoint 滚动窗口限流
    (Flask-Limiter 在测试 client 下不生效,故手动实现;多进程需切换 Redis)
  - login / reset-request / reset-confirm 三个路由应用装饰器

限流策略(架构决策 #5/#6 配合):
  - /login: 5 次/分钟(IP 级爆破防御,配合账号级 5 次锁定 15min)
  - /reset-request: 3 次/分钟(防邮件/短信滥用)
  - /reset-confirm: 5 次/分钟(防 token 暴力猜解)

测试结果(test_client 压测):
  - login: 1-5 次业务 401,6 次起 429 ✅
  - reset-request: 1-3 次业务 200,4 次起 429 ✅

依赖:commit b002e92 的 7 路由蓝图。

BIZ-127 架构决策:
  ✓ #5 限流 5 次/分钟(IP 级,与账号级锁定 15min 双重防御)

Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
costcodev
2026-09-05 07:13:50 +08:00
parent b002e929dc
commit 44f46a0e40
3 changed files with 108 additions and 8 deletions
+87 -7
View File
@@ -52,6 +52,88 @@ from auth import (
auth_bp = Blueprint('auth_api', __name__, url_prefix='/api/auth')
# ============================================================
# 限流配置(commit 2 整合 Flask-Limiter)
# ============================================================
# 限流维度:按 IP + 路径 组合
# - /login: 5 次/分钟(防爆破;5 次错就锁 15min 是账号级,这里是 IP 级双重防御)
# - /reset-request: 3 次/分钟(防滥用邮件/短信)
# - 其他:默认 60 次/分钟(防止爬虫)
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
# limiter 实例由 app.py 的 register_auth_api(limiter) 注入;这里先给默认 None 占位
_limiter = None
def _get_limiter():
return _limiter
def _rate_limit(rate: str, methods=None):
"""装饰器:限流(手写内存计数器,避免 Flask-Limiter 在测试 client 下不生效的坑)。
实现:按 (IP, view_function_name) 为 key,滚动 60 秒窗口。
- 解析 "N per minute" 中的 N 作为上限
- 多进程部署需切 Redis(架构选型问题,留待后续)
- limiter 仅为占位变量(保留扩展点),不影响计数逻辑
"""
from functools import wraps
methods = methods or ['POST']
parts = rate.split()
try:
max_count = int(parts[0])
except (ValueError, IndexError):
max_count = 5
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
# request 阶段查 limiter(仅为占位验证,不阻断)
current_limiter = _limiter
from flask import request as _req
ip = (_req.headers.get('X-Forwarded-For') or _req.remote_addr or 'unknown').split(',')[0].strip()
key = f'{ip}:{f.__name__}'
import time
now = time.time()
bucket = _rl_state.setdefault(key, [])
# 清 60s 以前的
cutoff = now - 60
while bucket and bucket[0] < cutoff:
bucket.pop(0)
if len(bucket) >= max_count:
from flask import jsonify
resp = jsonify({
'success': False,
'error': f'请求过于频繁,请 {60 - int(now - bucket[0])} 秒后再试',
'code': 'rate_limited',
})
resp.status_code = 429
return resp
bucket.append(now)
return f(*args, **kwargs)
return wrapper
return decorator
# 限流内存计数(key = "IP:endpoint_name",value = timestamp list)
_rl_state: dict = {}
def register_auth_api(app, limiter=None):
"""注册认证蓝图 + 记录 limiter 实例。
Args:
app: Flask app
limiter: 可选 flask_limiter.Limiter 实例。仅作为“能力标记”记录;
实际限流由 _rate_limit 装饰器手动实现(见 _rl_state)。
"""
global _limiter
_limiter = limiter
app.register_blueprint(auth_bp)
# ============================================================
# 工具函数
@@ -128,6 +210,7 @@ def require_session(f):
# ============================================================
@auth_bp.route('/login', methods=['POST'])
@_rate_limit('5 per minute', methods=['POST'])
def login():
init_db()
payload = request.get_json(silent=True) or {}
@@ -250,6 +333,7 @@ def change_password():
# ============================================================
@auth_bp.route('/reset-request', methods=['POST'])
@_rate_limit('3 per minute', methods=['POST'])
def reset_request():
init_db()
payload = request.get_json(silent=True) or {}
@@ -294,6 +378,7 @@ def reset_request():
# ============================================================
@auth_bp.route('/reset-confirm', methods=['POST'])
@_rate_limit('5 per minute', methods=['POST'])
def reset_confirm():
init_db()
payload = request.get_json(silent=True) or {}
@@ -355,10 +440,5 @@ def refresh():
return _ok({'refreshed': True, 'expires_at': new_expires})
# ============================================================
# 蓝图注册入口(供 app.py 调用)
# ============================================================
def register_auth_api(app):
"""在 app.py 里 app.register_blueprint(auth_bp) 即可。"""
app.register_blueprint(auth_bp)
# 蓝图注册入口已定义在文件顶部 register_auth_api(app, limiter=None)
# 避免重复定义(早期版本残留已清理)