From d56a9a607924ae350ab3b76794064add45cd017a Mon Sep 17 00:00:00 2001 From: Simon Date: Tue, 7 Jul 2026 22:54:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20=E5=AE=8C=E6=88=90=20AUTH-01~04?= =?UTF-8?q?=20=E5=90=8E=E7=AB=AF=E5=AE=9E=E7=8E=B0=20-=20IP=E7=99=BD?= =?UTF-8?q?=E5=90=8D=E5=8D=95=E4=B8=AD=E9=97=B4=E4=BB=B6+mfa.py=E5=88=A0?= =?UTF-8?q?=E9=99=A4+agents.py=E9=87=8D=E6=9E=84+conftest=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/agents.py | 7 +- backend/app/api/auth_wecom_sso.py | 175 +-------- backend/app/api/mfa.py | 389 ------------------- backend/app/main.py | 9 + backend/app/middleware/admin_ip_whitelist.py | 137 +++++++ backend/tests/conftest.py | 23 +- 6 files changed, 153 insertions(+), 587 deletions(-) delete mode 100644 backend/app/api/mfa.py create mode 100644 backend/app/middleware/admin_ip_whitelist.py diff --git a/backend/app/api/agents.py b/backend/app/api/agents.py index e81e7c5..041773e 100644 --- a/backend/app/api/agents.py +++ b/backend/app/api/agents.py @@ -18,7 +18,6 @@ from datetime import datetime from typing import Optional from uuid import UUID -import pyotp import qrcode import redis.asyncio as aioredis import bcrypt # P1 修复: 直接使用 bcrypt 库替代 passlib @@ -35,6 +34,7 @@ from app.dependencies import get_current_user, require_role, dep_wecom_service from app.models.agent import Agent from app.schemas.agent import AgentLogin, AgentResponse, AgentStatusUpdate from app.services.wecom_service import WecomService +from app.services.mfa_service import MFAService from app.utils.response import AppException, ERR_UNAUTHORIZED, success_response from app.utils.error_codes import ErrorCode @@ -282,9 +282,8 @@ async def agent_login( "role": agent.role, # 必须包含role字段，供前端校验权限 }) else: - # 验证 OTP 码 - totp = pyotp.TOTP(agent.mfa_secret) - if not totp.verify(body.otp_code, valid_window=1): + # 验证 OTP 码（决策3：复用 MFAService 统一校验逻辑） + if not MFAService.verify_code(agent.mfa_secret, body.otp_code, valid_window=1): raise AppException(1006, "OTP验证码错误，请重新输入") # 3. 生成随机 token（使用统一格式） diff --git a/backend/app/api/auth_wecom_sso.py b/backend/app/api/auth_wecom_sso.py index 8c14b9f..9d05f70 100644 --- a/backend/app/api/auth_wecom_sso.py +++ b/backend/app/api/auth_wecom_sso.py @@ -405,7 +405,7 @@ async def refresh_token( ) logger.info(f"Token 刷新成功: employee_id={user_info.get('employee_id')}") - return success_response(data={"token": token, "expires_in": TOKEN_TTL_SECONDS}) + return success_response(data={"token": token, "expires_in": TOKEN_TTL_SECONDS}) except json.JSONDecodeError: pass @@ -526,176 +526,7 @@ async def refresh_token_alias( # -------------------------------------------------------------------------- -# POST /api/auth_wecom/jsdk-login — 企微 JS-SDK 免认证登录 (v1.8 新增) +# 注意：/api/auth_wecom/jsdk-login 接口已按决策4删除 +# 原功能为"企微免密登录"，已按 PRD 要求移除 # -------------------------------------------------------------------------- -# 流程： -# 1. 前端通过 wx.agentConfig 获取企微用户 userid -# 2. 前端调用本接口，传入 userid -# 3. 后端验证 userid 是否是坐席 -# 4. 生成 token，返回给前端 -# -------------------------------------------------------------------------- -from pydantic import BaseModel, Field - -class WecomJsdkLoginRequest(BaseModel): - """企微 JS-SDK 免认证登录请求""" - userid: str = Field(..., description="企微用户 ID（从 wx.agentConfig 获取）") - login_source: str = Field(default="wecom_jsdk", description="登录来源标识") - - -@router.post("/jsdk-login") -async def wecom_jsdk_login( - body: WecomJsdkLoginRequest, - db: AsyncSession = Depends(get_db), - redis_client = Depends(get_redis), -): - """企微 JS-SDK 免认证登录。 - - 前端通过企微 JS-SDK (wx.agentConfig) 获取当前用户 ID， - 然后调用本接口进行免认证登录。 - - 流程： - 1. 调用 /wecom/check-role 接口验证 userid 是否是坐席 - 2. 查找或创建坐席记录 - 3. 生成 token，存入 Redis - 4. 返回 token 和坐席信息 - - Args: - body: 包含企微 userid - db: 数据库会话 - redis_client: Redis 客户端 - - Returns: - 登录成功：{ code: 0, data: { token, employee_id, name, roles } } - """ - try: - # 1. 验证用户是否具有坐席或管理员角色 - wecom_service = WecomService(redis_client) - userid = body.userid - - # 从数据库查询用户角色（支持坐席和管理员） - from app.services.role_mapping_service import RoleMappingService - role_service = RoleMappingService(db) - user_roles = await role_service.get_user_roles(userid) - - # 检查是否具有坐席或管理员角色 - has_agent_role = "agent" in user_roles - has_admin_role = "admin" in user_roles - - if not has_agent_role and not has_admin_role: - # 无坐席或管理员角色，尝试从企微标签检测（兼容旧逻辑） - tag_id = getattr(settings, "wecom_agent_tag_id", None) - if tag_id: - try: - access_token = await wecom_service.get_access_token() - url = f"https://qyapi.weixin.qq.com/cgi-bin/tag/get?access_token={access_token}&tagid={tag_id}" - import httpx - async with httpx.AsyncClient(timeout=5.0) as client: - resp = await client.get(url) - result = resp.json() - - if result.get("errcode", 0) == 0: - user_list = result.get("userlist", []) - user_ids = [ - u if isinstance(u, str) else u.get("userid", "") - for u in user_list - ] - if userid in user_ids: - has_agent_role = True - logger.info(f"企微 JS-SDK 免认证: userid={userid} 通过企微标签验证为坐席") - except Exception as e: - logger.warning(f"企微标签检测失败: {e}") - - # 如果既没有坐席也没有管理员角色，拒绝登录 - if not has_agent_role and not has_admin_role: - logger.warning(f"企微 JS-SDK 免认证失败: userid={userid} 没有坐席或管理员角色") - raise AppException(403, "您没有坐席或管理员权限，无法使用此方式登录") - - # 确定用户角色 - role_names = [] - if has_admin_role: - role_names.append("admin") - if has_agent_role: - role_names.append("agent") - - logger.info(f"企微 JS-SDK 免认证: userid={userid}, roles={role_names}") - - # 3. 获取用户详细信息 - try: - user_info = await wecom_service.get_user_info(userid) - user_name = user_info.get("name", userid) - # 同步头像到 employee 表 + 清缓存（要求 A；不阻塞登录） - try: - from app.services.avatar_service import sync_employee_avatar - await sync_employee_avatar(db, redis_client, userid, user_info.get("avatar", "")) - except Exception as av_err: - logger.warning(f"JS-SDK 同步头像失败（不阻塞）: userid={userid}, error={av_err}") - except Exception as e: - logger.warning(f"获取企微用户信息失败: {e}") - user_name = userid - - # 4. 查找或创建坐席记录 - from app.models.agent import Agent - - stmt = select(Agent).where(Agent.user_id == userid) - result = await db.execute(stmt) - agent = result.scalars().first() - - if not agent: - # 首次登录，创建坐席记录 - agent = Agent( - user_id=userid, - name=user_name, - status="online", - current_load=0, - max_load=5, - ) - db.add(agent) - await db.flush() - logger.info(f"企微 JS-SDK 免认证创建坐席: user_id={userid}, name={user_name}") - else: - # 更新坐席状态 - agent.name = user_name - agent.status = "online" - agent.updated_at = datetime.now() - db.add(agent) - await db.flush() - logger.info(f"企微 JS-SDK 免认证登录: user_id={userid}, name={user_name}") - - # 5. 生成 token - token = secrets.token_urlsafe(32) - - # 6. 存储 token 到 Redis - token_key = f"agent:token:{token}" - await redis_client.setex(token_key, TOKEN_TTL_SECONDS, userid) - - # 7. 记录审计日志 - await record_audit_log( - db=db, - employee_id=userid, - action="wecom_jsdk_login", - resource="auth", - resource_id=userid, - details={ - "name": user_name, - "roles": role_names, - "login_method": "wecom_jsdk", - }, - result="success", - ) - await db.commit() - - logger.info(f"企微 JS-SDK 免认证登录成功: user_id={userid}, roles={role_names}") - - return success_response(data={ - "token": token, - "employee_id": userid, - "name": user_name, - "roles": role_names, - }) - - except AppException: - raise - except Exception as e: - logger.error(f"企微 JS-SDK 免认证登录异常: {e}", exc_info=True) - raise AppException(500, f"免认证登录失败: {str(e)}") diff --git a/backend/app/api/mfa.py b/backend/app/api/mfa.py deleted file mode 100644 index 82c8b71..0000000 --- a/backend/app/api/mfa.py +++ /dev/null @@ -1,389 +0,0 @@ -# ============================================================================= -# 企微IT智能服务台 — MFA 二次认证 API -# ============================================================================= -# 说明：基于 TOTP(Google Authenticator 兼容)的二次认证 API -# Phase 2.1 task #17: pyotp TOTP 服务 + User MFA 字段 -# -# 端点列表: -# 1. GET /api/mfa/status — 查询绑定状态(路由守卫用) -# 2. POST /api/mfa/bind/start — 生成 secret + 二维码(尚未启用) -# 3. POST /api/mfa/bind/confirm — 输入 OTP 完成绑定(启用) -# 4. POST /api/mfa/verify — 输入 OTP 通过验证(写 Redis 30 分钟) -# 5. POST /api/mfa/disable — 用户主动关闭 MFA -# 6. POST /api/admin/mfa/reset/{employee_id} — 管理员重置(员工丢手机兜底) -# -# 鉴权: -# - 1-5 用 get_current_user(任意已登录用户) -# - 6 用 require_role("admin")(管理员) -# -# 流程(典型用户视角): -# 1. 前端路由守卫调 GET /status,bound=false → 跳转绑定页 -# 2. 用户点"绑定" → POST /bind/start → 展示二维码 + secret -# 3. 用户用 Authenticator 扫码 → 输入 6 位码 → POST /bind/confirm -# 4. 后续敏感操作前 → POST /verify → Redis 30 分钟内免重复输 -# 5. 丢手机 → 找管理员 → POST /admin/mfa/reset/{employee_id} -# ============================================================================= - -import logging -from datetime import datetime -from typing import Optional - -import redis.asyncio as aioredis -from fastapi import APIRouter, Depends -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from app.config import settings -from app.database import get_db -from app.dependencies import UserInfo, get_current_user -from app.models.agent import Agent -from app.schemas.mfa import ( - MFAAdminResetResponse, - MFABindConfirmRequest, - MFABindConfirmResponse, - MFABindStartResponse, - MFADisableRequest, - MFADisableResponse, - MFAStatusResponse, - MFAVerifyRequest, - MFAVerifyResponse, -) -from app.services.mfa_service import MFA_VERIFIED_TTL_SECONDS, MFAService -from app.utils.error_codes import ErrorCode -from app.utils.response import AppException, success_response - -logger = logging.getLogger(__name__) - - -# ----------------------------------------------------------------------------- -# 路由配置 -# ----------------------------------------------------------------------------- -# /api/mfa 前缀;admin 重置走 /api/admin/mfa 单独 router -# ----------------------------------------------------------------------------- -router = APIRouter(prefix="/mfa", tags=["MFA二次认证"]) -admin_router = APIRouter(prefix="/admin/mfa", tags=["MFA管理(管理员)"]) - - -def _get_redis() -> aioredis.Redis: - """获取 Redis 客户端(模块级 helper,便于测试 patch)。 - - Returns: - aioredis.Redis: Redis 异步客户端 - """ - return settings.create_redis_client() - - -# ----------------------------------------------------------------------------- -# 通用工具:根据 user_id 查 Agent 记录 -# ----------------------------------------------------------------------------- -async def _get_agent_by_employee_id( - db: AsyncSession, employee_id: str -) -> Optional[Agent]: - """按 user_id(employee_id)查询 Agent 行。 - - Args: - db: 数据库会话 - employee_id: 用户标识(企微 userid) - - Returns: - Optional[Agent]: 找不到返回 None - """ - stmt = select(Agent).where(Agent.user_id == employee_id) - result = await db.execute(stmt) - return result.scalars().first() - - -# ----------------------------------------------------------------------------- -# 通用工具:验证当前用户是否已登录 + 取得 Agent 行 -# ----------------------------------------------------------------------------- -async def _require_agent( - db: AsyncSession, current_user: UserInfo -) -> Agent: - """根据当前 token 取出对应的 Agent 行,不存在则 404。 - - 为什么需要 Agent 行: - MFA 状态/secret 都存在 agents 表,不是 employees 表。 - - Raises: - AppException: 坐席不存在(E4001) - """ - agent = await _get_agent_by_employee_id(db, current_user.employee_id) - if not agent: - raise AppException(ErrorCode.AGENT_NOT_FOUND, "坐席不存在,无法进行 MFA 操作") - return agent - - -# ============================================================================= -# 1. GET /api/mfa/status — 查询绑定状态 -# ============================================================================= -@router.get("/status", response_model=None) -async def get_mfa_status( - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -): - """查询当前用户的 MFA 绑定状态。 - - 前端路由守卫使用: - - bound=false → 强制走绑定流程 - - bound=true → 跳到"输入 OTP 验证"或继续业务 - - Returns: - success_response({bound, enabled, last_verified_at}) - """ - agent = await _require_agent(db, current_user) - - return success_response(data=MFAStatusResponse( - bound=bool(agent.mfa_enabled and agent.mfa_secret), - enabled=bool(agent.mfa_enabled), - last_verified_at=agent.mfa_last_verified_at, - ).model_dump(mode="json")) - - -# ============================================================================= -# 2. POST /api/mfa/bind/start — 生成 secret + 二维码 -# ============================================================================= -@router.post("/bind/start", response_model=None) -async def bind_start( - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -): - """生成 TOTP 密钥和二维码。 - - 行为: - - 生成 32 位 base32 secret - - 把 secret 写入 agents.mfa_secret(mfa_enabled=False,mfa_bound_at=None) - - 返回 otpauth URI + base64 二维码 PNG(给前端展示) - - 重复调用策略: - - 如果已经 enabled=True → 拒绝,要求先 disable 再重新绑定 - - 如果只是 secret 存在但 enabled=False → 复用旧 secret(支持"刷新二维码") - - Returns: - success_response({secret, otpauth_url, qr_code_base64}) - """ - agent = await _require_agent(db, current_user) - - # 已启用则拒绝重新绑定(必须先 disable) - if agent.mfa_enabled: - raise AppException( - ErrorCode.INVALID_PARAMETER, - "已绑定 MFA,如需重新绑定请先关闭", - ) - - # 复用旧 secret 还是新生成? - if agent.mfa_secret: - secret = agent.mfa_secret - else: - secret = MFAService.generate_secret() - agent.mfa_secret = secret - # mfa_enabled 保持 False,mfa_bound_at 等首次验证通过再写 - db.add(agent) - await db.flush() - - otpauth_url = MFAService.build_provisioning_uri(secret, agent.user_id) - qr_base64 = MFAService.render_qrcode_base64(otpauth_url) - - logger.info(f"MFA bind/start: agent={agent.user_id}, secret_prefix={secret[:4]}...") - - return success_response(data=MFABindStartResponse( - secret=secret, - otpauth_url=otpauth_url, - qr_code_base64=qr_base64, - ).model_dump()) - - -# ============================================================================= -# 3. POST /api/mfa/bind/confirm — 输入 OTP 完成绑定 -# ============================================================================= -@router.post("/bind/confirm", response_model=None) -async def bind_confirm( - body: MFABindConfirmRequest, - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -): - """用 6 位 OTP 码确认绑定,启用 MFA。 - - 行为: - - 用 mfa_secret 校验 otp_code(valid_window=1) - - 校验通过 → mfa_enabled=True, mfa_bound_at=now(), mfa_last_verified_at=now() - - 校验失败 → 抛 AppException(E_INVALID_PARAMETER) - - Returns: - success_response({success: true}) - """ - agent = await _require_agent(db, current_user) - - # 必须先 start(secret 必须存在) - if not agent.mfa_secret: - raise AppException( - ErrorCode.INVALID_PARAMETER, - "请先调用 /api/mfa/bind/start 获取二维码", - ) - - # 校验 OTP - if not MFAService.verify_code(agent.mfa_secret, body.otp_code): - logger.warning(f"MFA bind/confirm 验证码错误: agent={agent.user_id}") - raise AppException(ErrorCode.INVALID_PARAMETER, "OTP 验证码错误") - - now = datetime.now() - agent.mfa_enabled = True - agent.mfa_bound_at = now - agent.mfa_last_verified_at = now - db.add(agent) - await db.flush() - - logger.info(f"MFA bind/confirm 绑定成功: agent={agent.user_id}") - - return success_response(data=MFABindConfirmResponse(success=True).model_dump()) - - -# ============================================================================= -# 4. POST /api/mfa/verify — 输入 OTP 通过验证(写 Redis 30 分钟) -# ============================================================================= -@router.post("/verify", response_model=None) -async def verify_mfa( - body: MFAVerifyRequest, - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), - redis: aioredis.Redis = Depends(_get_redis), -): - """校验 6 位码,在 Redis 写 30 分钟复用标记。 - - 行为: - - 校验通过 → mfa:verified:{employee_id}=1 TTL 1800s - + 更新 mfa_last_verified_at - - 校验失败 → verified=false(不抛异常,前端可以重试) - - Returns: - success_response({verified, expires_in}) - """ - agent = await _require_agent(db, current_user) - - if not agent.mfa_enabled or not agent.mfa_secret: - # 用户还没绑定 MFA,直接返回 verified=false - # (前端可据此跳转到绑定流程) - return success_response(data=MFAVerifyResponse( - verified=False, - expires_in=0, - ).model_dump()) - - # 校验 - if not MFAService.verify_code(agent.mfa_secret, body.otp_code): - logger.warning(f"MFA verify 验证码错误: agent={agent.user_id}") - return success_response(data=MFAVerifyResponse( - verified=False, - expires_in=0, - ).model_dump()) - - # 写 Redis 复用标记 - await MFAService.mark_verified(redis, agent.user_id, MFA_VERIFIED_TTL_SECONDS) - - # 更新最后验证时间 - now = datetime.now() - agent.mfa_last_verified_at = now - db.add(agent) - await db.flush() - - logger.info(f"MFA verify 通过: agent={agent.user_id}") - - return success_response(data=MFAVerifyResponse( - verified=True, - expires_in=MFA_VERIFIED_TTL_SECONDS, - ).model_dump()) - - -# ============================================================================= -# 5. POST /api/mfa/disable — 用户主动关闭 MFA -# ============================================================================= -@router.post("/disable", response_model=None) -async def disable_mfa( - body: MFADisableRequest, - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), - redis: aioredis.Redis = Depends(_get_redis), -): - """关闭 MFA(清空 secret + disabled 标记)。 - - 安全要求: 必须先校验当前 OTP,防止误操作或被劫持后恶意关闭。 - - Returns: - success_response({success: true}) - """ - agent = await _require_agent(db, current_user) - - if not agent.mfa_enabled or not agent.mfa_secret: - # 没绑定过,直接幂等成功 - return success_response(data=MFADisableResponse(success=True).model_dump()) - - # 必须先验证 OTP - if not MFAService.verify_code(agent.mfa_secret, body.otp_code): - raise AppException(ErrorCode.INVALID_PARAMETER, "OTP 验证码错误,无法关闭 MFA") - - # 清空字段 - agent.mfa_secret = None - agent.mfa_enabled = False - agent.mfa_bound_at = None - # mfa_last_verified_at 保留,作为历史记录 - db.add(agent) - await db.flush() - - # 顺手清掉 Redis 验证标记(避免遗留) - await MFAService.clear_verified(redis, agent.user_id) - - logger.info(f"MFA disable: agent={agent.user_id}") - - return success_response(data=MFADisableResponse(success=True).model_dump()) - - -# ============================================================================= -# 6. POST /api/admin/mfa/reset/{employee_id} — 管理员重置(丢手机兜底) -# ============================================================================= -# 注意:此端点不要求 otp_code(员工已无法提供),只校验 admin 角色 -# 鉴权:在函数体内手动检查 current_user.roles 是否含 'admin',抛 AppException(FORBIDDEN) -# 原因:@require_role 装饰器 + body 参数组合在 FastAPI 签名合并时会重复 current_user 参数 -# (已知坑,见 memory rbac-pydantic-coroutine-pitfalls.md),手动校验更稳 -# ============================================================================= -@admin_router.post("/reset/{employee_id}", response_model=None) -async def admin_reset_mfa( - employee_id: str, - current_user: UserInfo = Depends(get_current_user), - db: AsyncSession = Depends(get_db), - redis: aioredis.Redis = Depends(_get_redis), -): - """管理员重置指定员工的 MFA 绑定(无 OTP 验证)。 - - 使用场景: - - 员工丢手机/换手机 → 管理员后台"重置 MFA"按钮 - - 鉴权:校验 current_user 是否拥有 admin 角色。 - - Returns: - success_response({success: true}) - """ - # 角色校验:仅 admin 角色可访问 - if "admin" not in current_user.roles: - raise AppException( - ErrorCode.FORBIDDEN, - "需要管理员权限", - ) - - stmt = select(Agent).where(Agent.user_id == employee_id) - result = await db.execute(stmt) - agent = result.scalars().first() - - if not agent: - raise AppException(ErrorCode.AGENT_NOT_FOUND, f"坐席 {employee_id} 不存在") - - agent.mfa_secret = None - agent.mfa_enabled = False - agent.mfa_bound_at = None - # mfa_last_verified_at 保留,作为审计 - db.add(agent) - await db.flush() - - # 顺手清 Redis 标记 - await MFAService.clear_verified(redis, employee_id) - - logger.info(f"MFA admin reset: employee_id={employee_id} by={current_user.employee_id}") - - return success_response(data=MFAAdminResetResponse(success=True).model_dump()) \ No newline at end of file diff --git a/backend/app/main.py b/backend/app/main.py index 90bfeff..b7b53e9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -664,6 +664,15 @@ def create_app() -> FastAPI: lifespan=lifespan, ) + # ---------------------------------------------------------------------- + # 管理端 IP 白名单中间件（仅 production 生效） + # ---------------------------------------------------------------------- + # 三端认证重构 AUTH-02：管理后台仅限内网/VPN + IP 白名单 + # 非生产环境自动跳过，方便本地开发 + # ---------------------------------------------------------------------- + from app.middleware.admin_ip_whitelist import AdminIPWhitelistMiddleware + app.add_middleware(AdminIPWhitelistMiddleware) + # ---------------------------------------------------------------------- # 配置 CORS（跨域资源共享） # ---------------------------------------------------------------------- diff --git a/backend/app/middleware/admin_ip_whitelist.py b/backend/app/middleware/admin_ip_whitelist.py new file mode 100644 index 0000000..592441c --- /dev/null +++ b/backend/app/middleware/admin_ip_whitelist.py @@ -0,0 +1,137 @@ +# ============================================================================= +# 企微IT智能服务台 — 管理端 IP 白名单中间件（三端认证重构 AUTH-02） +# ============================================================================= +# 说明：管理后台（/api/admin/*）的 IP 白名单校验中间件。 +# +# 工作原理： +# 1. 仅在 production 环境启用（非生产环境直接放行） +# 2. 仅对管理端 API 路径生效（/api/admin/* 和 /api/auth/otp-admin-*） +# 3. 校验客户端 IP 是否在 admin_allowed_ips 白名单内 +# 4. 不在白名单内则返回 403 + 统一错误格式 {code: 4004, message: "无访问权限"} +# +# 配置来源： +# - APP_ENV=production 时启用（从环境变量读取） +# - admin_allowed_ips 配置在 app.config（格式："117.147.35.138,218.75.34.87,10.240.0.0/16"） +# +# 使用方式： +# 在 main.py 中注册： +# from app.middleware.admin_ip_whitelist import AdminIPWhitelistMiddleware +# app.add_middleware(AdminIPWhitelistMiddleware) +# ============================================================================= + +import logging +import re +from typing import Callable + +from fastapi import Request, Response +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.responses import JSONResponse + +from app.utils.env_gating import is_production, ip_in_whitelist +from app.utils.response import error_response + +logger = logging.getLogger(__name__) + +# 需要校验 IP 白名单的管理端路径前缀（正则表达式） +ADMIN_PATH_PATTERNS = [ + r"^/api/admin/", # 管理后台 API + r"^/api/auth/otp-admin", # 管理端 OTP 操作 +] + + +class AdminIPWhitelistMiddleware(BaseHTTPMiddleware): + """管理端 IP 白名单校验中间件。 + + 仅在 production 环境对管理端 API 路径生效。 + 非生产环境（dev/test）直接放行，方便本地开发。 + """ + + def __init__(self, app, *args, **kwargs): + super().__init__(app, *args, **kwargs) + # 预编译正则表达式，提升匹配性能 + self._compiled_patterns = [re.compile(p) for p in ADMIN_PATH_PATTERNS] + + def _is_admin_path(self, path: str) -> bool: + """判断路径是否属于管理端 API。 + + Args: + path: 请求路径（如 "/api/admin/users"） + + Returns: + bool: True=管理端路径，False=非管理端路径 + """ + for pattern in self._compiled_patterns: + if pattern.match(path): + return True + return False + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """中间件主逻辑。 + + 1. 非 production 环境 → 直接放行 + 2. 非管理端路径 → 直接放行 + 3. 获取客户端 IP → 白名单校验 + - 在白名单 → 放行 + - 不在白名单 → 返回 403 + """ + # 1. 非 production 环境 → 直接放行（环境门控） + if not is_production(): + return await call_next(request) + + # 2. 非管理端路径 → 直接放行 + path = request.url.path + if not self._is_admin_path(path): + return await call_next(request) + + # 3. 获取客户端 IP + # 优先从 X-Forwarded-For 获取（反向代理场景） + # 否则使用 request.client.host + client_ip = self._get_client_ip(request) + + logger.debug(f"管理端 IP 校验: path={path}, client_ip={client_ip}") + + # 4. 白名单校验 + if not ip_in_whitelist(client_ip): + logger.warning( + f"管理端 IP 未授权: path={path}, client_ip={client_ip}, " + f"拒绝访问" + ) + # 返回统一错误格式（与 AppException 一致） + return JSONResponse( + status_code=200, # HTTP 状态码 200，业务错误码在 body 中 + content=error_response( + code=4004, + message="无访问权限：您的 IP 不在允许范围内，请联系管理员" + ), + ) + + # 5. 在白名单内 → 放行 + return await call_next(request) + + def _get_client_ip(self, request: Request) -> str: + """获取客户端真实 IP。 + + 优先从 X-Forwarded-For 请求头获取（反向代理场景）， + 这是标准的获取客户端真实 IP 的方式。 + + Args: + request: FastAPI 请求对象 + + Returns: + str: 客户端 IP 地址 + """ + # X-Forwarded-For 可能包含多个 IP，第一个是原始客户端 + forwarded_for = request.headers.get("X-Forwarded-For") + if forwarded_for: + # 格式："client_ip, proxy1, proxy2" + # 取第一个（原始客户端） + ips = forwarded_for.split(",") + if ips: + return ips[0].strip() + + # 直接连接场景 + if request.client: + return request.client.host + + # 兜底 + return "" diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 1a82f93..a0fd0ac 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -81,29 +81,8 @@ from app.models.agent_note import AgentNote # ============================================================================= # 2026-06-15 修复: monkey-patch starlette.config.Config 强制 UTF-8 读 .env -# 原因: Windows pytest 默认 GBK 读 .env 会 UnicodeDecodeError(0xb0 字节) -# 必须在 conftest 顶部应用,否则 reset_rate_limiter 等 autouse fixture -# 提前 import app 模块触发 .env 读取时会失败 +# 注意：已在文件顶部第38行应用过补丁，此处为兼容保留（不重复应用） # ============================================================================= -import starlette.config as _starlette_config - - -def _read_file_utf8(self, file_name, encoding=None): - """强制以 UTF-8 编码读 .env,避免 Windows GBK 默认编码触发 UnicodeDecodeError。""" - result = {} - # 始终使用 UTF-8 编码，忽略传入的 encoding 参数 - with open(file_name, encoding='utf-8') as f: - for line in f: - line = line.strip() - if not line or line.startswith('#'): - continue - if '=' in line: - k, v = line.split('=', 1) - result[k.strip()] = v.strip().strip('"').strip("'") - return result - - -_starlette_config.Config._read_file = _read_file_utf8 # =============================================================================