fix(auth): 登录速率限制与用户会话隔离增强

- auth_routes: 新增内存级登录速率限制(5min/10次),防止暴力破解
- auth_routes: 统一 DEV_MODE 判断收归 config.py 集中管理
- auth_routes: update_user 增加用户存在性校验
- gateway: require_role 装饰器恢复实际角色校验(原为占位实现)
- session_routes: get_user_sessions 统一传入 limit=20 参数
- session: get_history SQL 增加 id DESC 排序,修复同秒消息顺序错乱

🤖 Generated with [Qoder][https://qoder.com]
This commit is contained in:
lacerate551
2026-06-08 15:44:37 +08:00
parent 8adb550931
commit 3c262b8d35
4 changed files with 103 additions and 34 deletions

View File

@@ -12,7 +12,8 @@ from flask import Blueprint, request, jsonify
from auth.gateway import require_gateway_auth, require_role, get_user_permissions, MOCK_USERS
from core.status_codes import SUCCESS, BAD_REQUEST, UNAUTHORIZED, FORBIDDEN, PERMISSION_DENIED, INTERNAL_ERROR
from api.response_utils import success_response, error_response
import os
from config import DEV_MODE
import time
from pathlib import Path
from dotenv import load_dotenv
@@ -23,6 +24,37 @@ load_dotenv(env_path)
auth_bp = Blueprint('auth', __name__)
# ==================== 登录速率限制 ====================
# 简单的内存级速率限制:每个 IP 在时间窗口内最多允许 N 次登录尝试
_login_attempts = {} # {ip: [(timestamp, ...), ...]}
_RATE_LIMIT_WINDOW = 300 # 5 分钟窗口
_RATE_LIMIT_MAX = 10 # 窗口内最多 10 次尝试
def _check_rate_limit(client_ip: str) -> bool:
"""检查是否超出登录速率限制,返回 True 表示允许"""
now = time.time()
if client_ip not in _login_attempts:
_login_attempts[client_ip] = []
# 清理过期记录
_login_attempts[client_ip] = [
t for t in _login_attempts[client_ip]
if now - t < _RATE_LIMIT_WINDOW
]
if len(_login_attempts[client_ip]) >= _RATE_LIMIT_MAX:
return False
_login_attempts[client_ip].append(now)
return True
def _is_dev_mode() -> bool:
"""统一的开发模式判断(由 config.py 集中管理)"""
return DEV_MODE
@auth_bp.route('/auth/login', methods=['POST'])
def mock_login():
"""
@@ -52,9 +84,17 @@ def mock_login():
- manager / manager123 (经理,财务部)
- user / test123 (普通用户,技术部)
"""
# 默认开启开发模式(生产环境需设置 DEV_MODE=false
if os.environ.get('DEV_MODE', 'true').lower() == 'false':
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用,请设置 DEV_MODE=true", http_status=403)
if not _is_dev_mode():
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用,请在 .env 中设置 DEV_MODE=true", http_status=403)
# 速率限制检查
client_ip = request.remote_addr or 'unknown'
if not _check_rate_limit(client_ip):
return error_response(
"RATE_LIMITED", FORBIDDEN,
f"登录尝试过于频繁,请 {_RATE_LIMIT_WINDOW // 60} 分钟后再试",
http_status=429
)
data = request.json or {}
username = data.get('username')
@@ -81,7 +121,9 @@ def mock_login():
def get_stats():
"""获取系统统计信息(仅管理员)"""
from flask import current_app
session_manager = current_app.config['SESSION_MANAGER']
session_manager = current_app.config.get('SESSION_MANAGER')
if not session_manager:
return error_response("UNAVAILABLE", INTERNAL_ERROR, "会话管理器未启用", http_status=503)
return success_response(data=session_manager.get_stats())
@@ -133,8 +175,7 @@ def get_users():
]
}
"""
dev_mode = os.environ.get('DEV_MODE', 'true').lower() != 'false'
if not dev_mode:
if not _is_dev_mode():
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
users = []
@@ -161,12 +202,26 @@ def update_user(user_id):
"is_active": false
}
"""
dev_mode = os.environ.get('DEV_MODE', 'true').lower() != 'false'
if not dev_mode:
if not _is_dev_mode():
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
# 模拟用户不支持真正的状态切换,直接返回成功
return success_response(data={"message": "操作成功(模拟)", "user_id": user_id})
# 验证目标用户是否存在
target_user = None
for username, info in MOCK_USERS.items():
if info['user_id'] == user_id:
target_user = info
break
if not target_user:
return error_response("NOT_FOUND", BAD_REQUEST, f"用户 {user_id} 不存在", http_status=404)
data = request.json or {}
# 模拟操作记录请求但不实际执行mock 用户数据是静态的)
return success_response(data={
"message": "操作成功(模拟)",
"user_id": user_id,
"applied_changes": data
})
@auth_bp.route('/auth/change-password', methods=['POST'])
@@ -181,8 +236,7 @@ def change_password():
"new_password": "xxx"
}
"""
dev_mode = os.environ.get('DEV_MODE', 'true').lower() != 'false'
if not dev_mode:
if not _is_dev_mode():
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
data = request.json or {}
@@ -195,5 +249,12 @@ def change_password():
if len(new_password) < 6:
return error_response("INVALID_PARAMS", BAD_REQUEST, "新密码至少6位", http_status=400)
# 模拟环境直接返回成功
# 验证当前用户的旧密码
user = request.current_user
username = user.get('username', '')
mock_user = MOCK_USERS.get(username)
if mock_user and mock_user['password'] != old_password:
return error_response("UNAUTHORIZED", UNAUTHORIZED, "旧密码错误", http_status=401)
# 模拟环境返回成功不实际修改密码mock 数据是静态的)
return success_response(message="密码修改成功(模拟)")