feat(auth): 完成 AUTH-01~04 后端实现 - IP白名单中间件+mfa.py删除+agents.py重构+conftest修复

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（使用统一格式）

backend/app/api/auth_wecom_sso.py

@@ -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)}")

backend/app/api/mfa.py

@@ -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())

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（跨域资源共享）
     # ----------------------------------------------------------------------

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 ""

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
 
 
 # =============================================================================