A组认证加固: P0兜底+Token刷新+环境检测+OTP+RBRAC落地+P1日志审计+Token撤销 - 全局一致性审查通过

backend/app/dependencies.py

@@ -12,16 +12,22 @@ import json
 import logging
 from dataclasses import dataclass
 from functools import wraps
-from typing import List, Optional
+from typing import List, Optional, Union
 
 import redis.asyncio as aioredis
 from fastapi import Depends, HTTPException, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from app.config import settings
+from app.models.agent import Agent
 from app.services.token_service import TokenService
 from app.utils.response import AppException
 
+# 延迟导入 get_current_agent 以避免循环依赖
+def _get_current_agent():
+    from app.api.agents import get_current_agent
+    return get_current_agent
+
 logger = logging.getLogger(__name__)
 
 # HTTP Bearer 认证方案
@@ -324,30 +330,63 @@ def require_permission(
     def decorator(func):
         sig = inspect.signature(func)
         params = list(sig.parameters.values())
-        params.append(
-            inspect.Parameter(
-                'current_user',
-                inspect.Parameter.KEYWORD_ONLY,
-                annotation=UserInfo,
-                default=Depends(get_current_user),
+        param_names = {p.name for p in params}
+
+        # 智能检测参数名：优先使用函数已定义的参数名
+        # 支持 current_user (通用/管理端) 和 current_agent (坐席端)
+        if 'current_agent' in param_names:
+            param_name = 'current_agent'
+            param_annotation = Agent
+            param_default = Depends(_get_current_agent())
+        else:
+            param_name = 'current_user'
+            param_annotation = UserInfo
+            param_default = Depends(get_current_user)
+
+        # 检查是否需要添加参数
+        needs_param = param_name not in param_names
+        if needs_param:
+            params.append(
+                inspect.Parameter(
+                    param_name,
+                    inspect.Parameter.KEYWORD_ONLY,
+                    annotation=param_annotation,
+                    default=param_default,
+                )
             )
-        )
         new_sig = sig.replace(parameters=params)
 
         @wraps(func)
         async def wrapper(*args, **kwargs):
-            current_user = kwargs.pop('current_user')
+            # 提取注入的用户/坐席信息
+            current_user = kwargs.pop(param_name)
 
             # 拉用户所有角色的 permissions
             # 注: UserInfo.roles 是角色名列表,permissions 是 {role: [perm]} 字典
             # 首次实现简化: 角色判断 + admin 通配符
             # 完整实现需要查 DB 拉 permissions,见 rbac_service.check_permission
 
-            user_roles = set(current_user.roles or [])
+            # 支持两种类型：
+            # 1. UserInfo (H5/管理端): 有 roles 属性 (List[str])
+            # 2. Agent (坐席端): 有 role 属性 (str)
+            if hasattr(current_user, 'roles'):
+                user_roles = set(current_user.roles or [])
+                user_id = current_user.employee_id
+            elif hasattr(current_user, 'role'):
+                # Agent 类型：role 是字符串，直接作为角色
+                user_roles = {current_user.role} if current_user.role else set()
+                user_id = current_user.user_id
+            else:
+                # 兼容：没有 roles 或 role 属性的情况
+                user_roles = set()
+                user_id = getattr(current_user, 'user_id', 'unknown')
+
+            # 保存 user_id 供后续使用
+            current_user._rbac_user_id = user_id
 
             # 1. admin 角色直通(通配符 *:*:all)
             if "admin" in user_roles:
-                return await func(*args, current_user=current_user, **kwargs)
+                return await func(*args, **{param_name: current_user}, **kwargs)
 
             # 2. 其他角色: 走 rbac_service.check_permission
             #    简化: 这里只看角色名,不查 DB(性能考虑)
@@ -372,7 +411,7 @@ def require_permission(
 
             if not has_perm:
                 logger.warning(
-                    f"用户 {current_user.employee_id} 权限不足: "
+                    f"用户 {user_id} 权限不足: "
                     f"角色 {list(user_roles)}, 缺 {perm_string}"
                 )
                 raise HTTPException(
@@ -380,7 +419,7 @@ def require_permission(
                     detail=f"权限不足: 需要 {perm_string}",
                 )
 
-            return await func(*args, current_user=current_user, **kwargs)
+            return await func(*args, **{param_name: current_user}, **kwargs)
 
         wrapper.__signature__ = new_sig
         return wrapper