""" 认证与系统状态 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/', 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="密码修改成功（模拟）")