# -*- coding: utf-8 -*- """ Web UI 认证相关的 FastAPI 依赖项。 定义了用于验证 JWT 令牌的依赖函数。 """ import logging # 导入日志模块 from typing import Optional, Dict, Any # 导入类型提示 from fastapi import Depends, HTTPException, status, Request # 导入 FastAPI 相关组件 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials # 导入用于处理 Bearer Token 的安全工具 # 导入 JWT 解码函数 from app.core.security.jwt import decode_access_token # (新路径) logger = logging.getLogger('my_logger') # 获取日志记录器实例 # 创建一个 HTTP Bearer scheme 实例 # HTTPBearer 是 FastAPI 提供的一个安全工具类,用于从请求的 Authorization 头中提取 Bearer Token。 # auto_error=False: 表示如果请求头中没有 Authorization 或格式不正确,FastAPI 不会自动抛出 401 错误。 # 我们将在依赖函数 `verify_jwt_token` 内部手动检查 `credentials` 是否为 None,并抛出更具体的错误。 # 这允许我们提供更详细的日志记录或自定义错误响应。 bearer_scheme = HTTPBearer(auto_error=False) async def verify_jwt_token( request: Request, # 注入 FastAPI 请求对象,虽然当前未使用,但保留可能用于未来扩展(如记录请求信息) credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme) # 依赖注入:尝试从请求头获取 Bearer Token 凭证 ) -> Dict[str, Any]: """ FastAPI 依赖项函数,用于验证通过 HTTP Bearer 方案提供的 JWT 令牌。 此函数会检查 Authorization 请求头,提取 Bearer Token,然后解码并验证它。 Args: request (Request): FastAPI 请求对象 (当前未使用)。 credentials (Optional[HTTPAuthorizationCredentials]): FastAPI 从 Authorization 头提取的凭证对象。 如果请求头不存在或格式不正确,此值为 None。 Returns: Dict[str, Any]: 如果令牌有效,返回解码后的 JWT payload (包含用户信息,如 'sub' 和 'admin')。 Raises: HTTPException: - 401 Unauthorized: 如果没有提供有效的 Bearer Token,或者 Token 无效、过期、或缺少必要信息。 """ logger.debug("verify_jwt_token: 开始验证 JWT token (仅从 Bearer Header 获取)") # 记录开始验证日志 token = None # 初始化 token 变量 # 检查 credentials 是否存在,并且 scheme 是否为 'bearer' (不区分大小写) if credentials and credentials.scheme.lower() == "bearer": logger.debug("verify_jwt_token: 从 Authorization 头中找到 Bearer token。") # 记录找到 Token 日志 token = credentials.credentials # 获取实际的 token 字符串 # 如果未能从请求头中获取到 token if not token: logger.warning("verify_jwt_token: 未找到有效的认证令牌 (仅检查 Bearer Header)。") # 记录警告日志 # 抛出 401 未授权错误 raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, # 设置状态码 detail="缺少有效的认证令牌", # 设置错误详情 headers={"WWW-Authenticate": "Bearer"}, # 添加响应头,指示需要 Bearer 认证 ) # 记录提取到的 token (部分隐藏) logger.debug(f"verify_jwt_token: 提取到 token (前10位): {token[:10]}...") # --- 解码并验证 token --- # 调用 jwt.py 中的 decode_access_token 函数 payload = decode_access_token(token) # 解码 token # 如果解码失败或 token 无效/过期,decode_access_token 会返回 None if payload is None: logger.warning(f"verify_jwt_token: JWT token 无效或已过期 (Token: {token[:10]}...)") # 记录警告日志 # 抛出 401 未授权错误 raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, # 设置状态码 detail="无效或已过期的 token", # 设置错误详情 headers={"WWW-Authenticate": "Bearer"}, # 添加响应头 ) # 记录解码成功的 payload (用于调试) logger.debug(f"verify_jwt_token: JWT payload 解码成功: {payload}") # --- 检查 payload 内容 --- # 验证 payload 中是否包含必要的用户信息 (例如 'sub' 字段) user_key = payload.get("sub") # 获取 'sub' 字段 (通常是用户标识符) is_admin = payload.get("admin", False) # 获取 'admin' 字段 (布尔值),默认为 False if not user_key: # 如果缺少 'sub' 字段 logger.warning(f"JWT token 有效,但缺少 'sub' (用户标识符) 字段。Payload: {payload}") # 记录警告日志 # 抛出 401 未授权错误 raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, # 设置状态码 detail="无效的 token (缺少用户信息)", # 设置错误详情 headers={"WWW-Authenticate": "Bearer"}, # 添加响应头 ) # 如果所有验证通过 logger.debug(f"verify_jwt_token: JWT token 验证成功. User Key: {user_key[:8]}..., Is Admin: {is_admin}") # 记录成功日志 # 返回解码后的 payload 字典,路由处理函数可以从中获取用户信息 return payload async def verify_jwt_token_optional( credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme) ) -> Optional[Dict[str, Any]]: """ FastAPI 依赖项函数,可选地验证通过 HTTP Bearer 方案提供的 JWT 令牌。 如果令牌有效,返回解码后的 payload;如果无效、过期或未提供,则返回 None。 Args: credentials (Optional[HTTPAuthorizationCredentials]): FastAPI 从 Authorization 头提取的凭证对象。 Returns: Optional[Dict[str, Any]]: 如果令牌有效,返回解码后的 JWT payload,否则返回 None。 """ logger.debug("verify_jwt_token_optional: 开始可选的 JWT token 验证") token = None if credentials and credentials.scheme.lower() == "bearer": token = credentials.credentials logger.debug(f"verify_jwt_token_optional: 提取到 token (前10位): {token[:10]}...") if not token: logger.debug("verify_jwt_token_optional: 未找到 Bearer token。") return None payload = decode_access_token(token) # decode_access_token 内部处理无效/过期情况并返回 None if payload is None: logger.debug(f"verify_jwt_token_optional: JWT token 无效或已过期 (Token: {token[:10]}...)") return None user_key = payload.get("sub") if not user_key: logger.warning(f"verify_jwt_token_optional: JWT token 有效,但缺少 'sub' 字段。Payload: {payload}") return None # 视为无效 logger.debug(f"verify_jwt_token_optional: JWT token 验证成功. User Key: {user_key[:8]}...") return payload