# -*- coding: utf-8 -*- """ 安全相关工具函数,主要用于 JWT (JSON Web Token) 令牌的创建和验证。 使用 python-jose 库进行 JWT 操作。 """ from datetime import datetime, timedelta, timezone # 导入日期时间处理相关模块 from typing import Optional, Dict, Any # 导入类型提示 import logging # 导入日志模块 from jose import jwt, JWTError # 从 jose 库导入 jwt 编码/解码函数和 JWT 错误类 from app import config # 导入应用配置模块 logger = logging.getLogger('my_logger') # 获取日志记录器实例 # --- 从配置中加载 JWT 相关设置 --- # SECRET_KEY: 用于签名和验证 JWT 的密钥。必须保密! SECRET_KEY = config.SECRET_KEY # ALGORITHM: 用于签名 JWT 的算法,例如 "HS256"。 ALGORITHM = config.JWT_ALGORITHM # ACCESS_TOKEN_EXPIRE_MINUTES: 访问令牌的默认过期时间(分钟)。 ACCESS_TOKEN_EXPIRE_MINUTES = config.ACCESS_TOKEN_EXPIRE_MINUTES def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str: """ 创建 JWT 访问令牌。 Args: data (Dict[str, Any]): 需要编码到令牌 payload 中的数据字典 (例如包含用户 ID 'sub')。 expires_delta (Optional[timedelta]): 可选参数,指定令牌的有效时长。 如果未提供,则使用配置中的 `ACCESS_TOKEN_EXPIRE_MINUTES`。 Returns: str: 编码后的 JWT 访问令牌字符串。 Raises: ValueError: 如果配置中未设置 `SECRET_KEY`。 Exception: 如果在 JWT 编码过程中发生其他错误。 """ # 检查密钥是否存在 if not SECRET_KEY: logger.error("JWT SECRET_KEY 未配置,无法创建令牌。") # 记录严重错误 raise ValueError("JWT SECRET_KEY 未在环境变量中设置") # 抛出值错误 to_encode = data.copy() # 创建数据字典的副本,避免修改原始数据 # 计算过期时间 if expires_delta: # 如果提供了自定义的过期时长 expire = datetime.now(timezone.utc) + expires_delta # 使用当前 UTC 时间加上指定时长 else: # 如果未提供自定义时长 # 使用配置中定义的默认分钟数计算过期时间 expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) # 将过期时间 ('exp' claim) 添加到要编码的数据中 to_encode.update({"exp": expire}) try: # 使用 jose.jwt.encode 函数生成 JWT # - to_encode: 要编码的 payload 数据 # - SECRET_KEY: 签名密钥 # - algorithm: 签名算法 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) logger.debug(f"成功创建 JWT 令牌,过期时间: {expire}") # 记录调试日志 return encoded_jwt # 返回编码后的 JWT 字符串 except Exception as e: # 捕获编码过程中可能发生的任何异常 logger.error(f"创建 JWT 时发生错误: {e}", exc_info=True) # 记录错误日志 raise # 重新抛出异常,让上层调用者处理 def decode_access_token(token: str) -> Optional[Dict[str, Any]]: """ 解码并验证 JWT 访问令牌。 验证内容包括:签名、过期时间。 Args: token (str): 需要解码和验证的 JWT 字符串。 Returns: Optional[Dict[str, Any]]: 如果令牌有效且未过期,返回解码后的 payload 字典; 否则返回 None。 Raises: ValueError: 如果配置中未设置 `SECRET_KEY`。 """ # 检查密钥是否存在 if not SECRET_KEY: logger.error("JWT SECRET_KEY 未配置,无法解码令牌。") # 记录严重错误 raise ValueError("JWT SECRET_KEY 未在环境变量中设置") # 抛出值错误 # 记录调试信息,显示使用的密钥前缀和算法 logger.debug(f"decode_access_token: 使用的 SECRET_KEY (前8位): {SECRET_KEY[:8]}...") logger.debug(f"decode_access_token: 使用的 ALGORITHM: {ALGORITHM}") try: # 使用 jose.jwt.decode 函数解码并验证 JWT # - token: 要解码的 JWT 字符串 # - SECRET_KEY: 用于验证签名的密钥 # - algorithms: 指定允许的签名算法列表 # decode 函数会自动验证签名和过期时间 ('exp' claim) payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) # --- 可选的额外验证 --- # 1. 检查 'exp' 字段是否存在且类型正确 (虽然 jose 库通常会处理) if "exp" not in payload or not isinstance(payload["exp"], (int, float)): logger.warning(f"解码后的 JWT payload 中缺少有效的 'exp' 字段: {payload}") # 记录警告 return None # 视为无效 # 2. 可以再次显式检查过期时间 (双重保险) # exp_timestamp = payload["exp"] # if datetime.now(timezone.utc).timestamp() > exp_timestamp: # logger.debug(f"JWT 令牌已过期 (显式检查): {token[:10]}...") # return None # 3. 可以在这里添加其他业务逻辑相关的验证,例如检查 payload 中是否包含必要的字段 ('sub' 等) # if "sub" not in payload: # logger.warning(f"解码后的 JWT payload 中缺少 'sub' 字段: {payload}") # return None # 如果所有验证通过,返回解码后的 payload return payload except JWTError as e: # 捕获 jose 库抛出的所有 JWT 相关错误 # JWTError 包括签名无效、令牌过期、格式错误等多种情况 logger.debug(f"JWT 验证失败: {e} (Token: {token[:10]}...)") # 记录 JWT 验证失败的调试信息,不暴露完整 token return None # 返回 None 表示令牌无效 except Exception as e: # 捕获解码过程中可能发生的其他意外错误 logger.error(f"解码 token 时发生意外错误: {e}", exc_info=True) # 记录错误日志 return None # 返回 None