3c262b8d35
- 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]
261 lines
7.6 KiB
Python
261 lines
7.6 KiB
Python
"""
|
||
认证与系统状态 API
|
||
|
||
路由:
|
||
- POST /auth/login - 模拟登录（仅开发环境）
|
||
- GET /stats - 系统统计 (管理员)
|
||
- GET /health - 健康检查
|
||
- GET /auth/me - 当前用户信息
|
||
"""
|
||
|
||
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
|
||
from config import DEV_MODE
|
||
import time
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载 .env 文件（从项目根目录）
|
||
env_path = Path(__file__).parent.parent / '.env'
|
||
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():
|
||
"""
|
||
模拟登录（仅开发环境）
|
||
|
||
请求体:
|
||
{
|
||
"username": "admin",
|
||
"password": "admin123"
|
||
}
|
||
|
||
返回:
|
||
{
|
||
"token": "mock-token-admin",
|
||
"user": {
|
||
"user_id": "admin001",
|
||
"username": "admin",
|
||
"role": "admin",
|
||
"department": "管理部"
|
||
}
|
||
}
|
||
|
||
测试账号:
|
||
- admin / admin123 (管理员，管理部)
|
||
- admin2 / admin456 (管理员，技术部)
|
||
- admin3 / admin789 (管理员，运营部)
|
||
- manager / manager123 (经理，财务部)
|
||
- user / test123 (普通用户，技术部)
|
||
"""
|
||
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')
|
||
password = data.get('password')
|
||
|
||
user = MOCK_USERS.get(username)
|
||
if not user or user['password'] != password:
|
||
return error_response("UNAUTHORIZED", UNAUTHORIZED, "用户名或密码错误", http_status=401)
|
||
|
||
return success_response(data={
|
||
"token": f"mock-token-{username}",
|
||
"user": {
|
||
"user_id": user['user_id'],
|
||
"username": username,
|
||
"role": user['role'],
|
||
"department": user['department']
|
||
}
|
||
})
|
||
|
||
|
||
@auth_bp.route('/stats', methods=['GET'])
|
||
@require_gateway_auth
|
||
@require_role('admin')
|
||
def get_stats():
|
||
"""获取系统统计信息（仅管理员）"""
|
||
from flask import current_app
|
||
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())
|
||
|
||
|
||
@auth_bp.route('/health', methods=['GET'])
|
||
def health():
|
||
"""健康检查"""
|
||
return jsonify({
|
||
"status": "ok",
|
||
"knowledge_base": "多向量库模式 (按集合提供服务)",
|
||
"bm25_index": "动态按需加载",
|
||
"mode": "Agentic RAG"
|
||
})
|
||
|
||
|
||
@auth_bp.route('/auth/me', methods=['GET'])
|
||
@require_gateway_auth
|
||
def get_current_user():
|
||
"""
|
||
获取当前用户信息
|
||
|
||
开发模式下支持模拟用户，生产模式下用户信息由后端控制。
|
||
"""
|
||
user = request.current_user
|
||
return success_response(data={
|
||
"user_id": user["user_id"],
|
||
"username": user["username"],
|
||
"role": user["role"],
|
||
"department": user["department"],
|
||
"permissions": get_user_permissions(user["role"])
|
||
})
|
||
|
||
|
||
@auth_bp.route('/auth/users', methods=['GET'])
|
||
@require_gateway_auth
|
||
def get_users():
|
||
"""
|
||
获取用户列表（仅开发环境）
|
||
|
||
返回:
|
||
{
|
||
"users": [
|
||
{
|
||
"user_id": "admin001",
|
||
"username": "admin",
|
||
"role": "admin",
|
||
"department": "管理部",
|
||
"is_active": true
|
||
}
|
||
]
|
||
}
|
||
"""
|
||
if not _is_dev_mode():
|
||
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
|
||
|
||
users = []
|
||
for username, info in MOCK_USERS.items():
|
||
users.append({
|
||
"user_id": info["user_id"],
|
||
"username": username,
|
||
"role": info["role"],
|
||
"department": info["department"],
|
||
"is_active": True # 模拟用户默认都是活跃状态
|
||
})
|
||
|
||
return success_response(data={"users": users})
|
||
|
||
|
||
@auth_bp.route('/auth/users/<user_id>', methods=['PUT'])
|
||
@require_gateway_auth
|
||
def update_user(user_id):
|
||
"""
|
||
更新用户信息（仅开发环境，模拟操作）
|
||
|
||
请求体:
|
||
{
|
||
"is_active": false
|
||
}
|
||
"""
|
||
if not _is_dev_mode():
|
||
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
|
||
|
||
# 验证目标用户是否存在
|
||
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'])
|
||
@require_gateway_auth
|
||
def change_password():
|
||
"""
|
||
修改密码（仅开发环境，模拟操作）
|
||
|
||
请求体:
|
||
{
|
||
"old_password": "xxx",
|
||
"new_password": "xxx"
|
||
}
|
||
"""
|
||
if not _is_dev_mode():
|
||
return error_response("FORBIDDEN", FORBIDDEN, "仅开发环境可用", http_status=403)
|
||
|
||
data = request.json or {}
|
||
old_password = data.get('old_password')
|
||
new_password = data.get('new_password')
|
||
|
||
if not old_password or not new_password:
|
||
return error_response("MISSING_PARAMS", BAD_REQUEST, "请提供旧密码和新密码", http_status=400)
|
||
|
||
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="密码修改成功（模拟）")
|