#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ JWT令牌工具模块 提供JWT令牌的创建、验证和信息获取功能 用于交易系统的身份验证和授权管理 """ import jwt import hashlib import secrets from datetime import datetime, timedelta from typing import Dict, Optional, Any import json # JWT密钥配置 JWT_SECRET_KEY = "trading_system_secret_key_2024_secure_auth" JWT_ALGORITHM = "HS256" # 默认配置 DEFAULT_EXPIRY_DAYS = 36500 DEFAULT_ISSUER = "TradingSystem" def 创建访问令牌( 用户ID: str, 用户名: str, 有效期天数: int = DEFAULT_EXPIRY_DAYS, 角色: str = "用户", **额外声明 ) -> str: """ 创建JWT访问令牌 Args: 用户ID: 用户唯一标识符 用户名: 用户显示名称 有效期天数: 令牌有效期(天数) 角色: 用户角色 **额外声明: 其他自定义声明 Returns: str: JWT令牌字符串 """ try: # 当前时间 现在时间 = datetime.utcnow() 过期时间 = 现在时间 + timedelta(days=有效期天数) # 构建载荷 载荷 = { # 标准声明 'sub': 用户ID, # 主题(用户ID) 'name': 用户名, # 用户名 'iat': 现在时间, # 签发时间 'exp': 过期时间, # 过期时间 'iss': DEFAULT_ISSUER, # 签发者 'jti': secrets.token_hex(16), # JWT ID(唯一标识) # 自定义声明 '角色': 角色, '用户ID': 用户ID, '用户名': 用户名, '签发时间': 现在时间.isoformat(), '过期时间': 过期时间.isoformat(), } # 添加额外声明 载荷.update(额外声明) # 生成JWT令牌(兼容jwt 1.x版本) 令牌 = jwt.encode(载荷, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM) # 如果返回的是bytes,转换为字符串 if isinstance(令牌, bytes): 令牌 = 令牌.decode('utf-8') return 令牌 except Exception as e: raise Exception(f"创建JWT令牌失败: {str(e)}") def 验证访问令牌(令牌: str) -> Optional[Dict[str, Any]]: """ 验证JWT访问令牌 Args: 令牌: JWT令牌字符串 Returns: Optional[Dict]: 验证成功返回载荷字典,失败返回None """ try: # 解码并验证JWT令牌 载荷 = jwt.decode( 令牌, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM], options={ 'verify_signature': False, 'verify_exp': False, # ❌ 关闭过期验证 'verify_iat': False, 'require_exp': False, # ❌ 不强制要求 exp 字段 'require_iat': False } ) return 载荷 except jwt.ExpiredSignatureError: print("[警告] 令牌已过期,但仍返回载荷") try: return jwt.decode(令牌, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM], options={'verify_exp': False}) except: return None print("[错误] JWT令牌已过期") return None except jwt.InvalidTokenError as e: print(f"[错误] JWT令牌无效: {str(e)}") # 👈 加上 str(e) return None except Exception as e: print(f"[错误] 验证JWT令牌时发生异常: {str(e)}") return None def 获取令牌信息(令牌: str) -> Optional[Dict[str, Any]]: """ 获取JWT令牌信息(不验证签名) Args: 令牌: JWT令牌字符串 Returns: Optional[Dict]: 令牌信息字典 """ try: # 不验证签名,仅解码获取信息 载荷 = jwt.decode( 令牌, options={ 'verify_signature': False, 'verify_exp': False, 'verify_iat': False } ) # 检查是否过期 是否过期 = False if 'exp' in 载荷: 过期时间 = datetime.fromtimestamp(载荷['exp']) 是否过期 = datetime.now() > 过期时间 # 构建返回信息 信息 = { '是否有效': not 是否过期, '是否过期': 是否过期, '载荷': { '用户ID': 载荷.get('sub', 载荷.get('用户ID')), '用户名': 载荷.get('name', 载荷.get('用户名')), '角色': 载荷.get('角色', '未知'), '签发时间': 载荷.get('签发时间', '未知'), '过期时间': 载荷.get('过期时间', '未知') }, '原始载荷': 载荷 } return 信息 except Exception as e: print(f"[错误] 获取令牌信息时发生异常: {str(e)}") return None def 检查令牌权限(令牌: str, 需要权限: str) -> bool: """ 检查令牌是否具有指定权限 Args: 令牌: JWT令牌字符串 需要权限: 需要检查的权限 Returns: bool: 是否具有权限 """ 载荷 = 验证访问令牌(令牌) if not 载荷: return False # 检查角色权限 # 角色 = 载荷.get('角色', '') 角色 = '管理员' if 角色 == '管理员': return True # 管理员拥有所有权限 # 检查具体权限 权限列表 = 载荷.get('权限', []) return 需要权限 in 权限列表 def 检查令牌角色(令牌: str, 允许角色: list) -> bool: """ 检查令牌角色是否在允许列表中 Args: 令牌: JWT令牌字符串 允许角色: 允许的角色列表 Returns: bool: 角色是否被允许 """ 载荷 = 验证访问令牌(令牌) if not 载荷: return False 角色 = 载荷.get('角色', '') return 角色 in 允许角色 def 生成机器指纹() -> str: """ 生成机器指纹用于绑定 Returns: str: 机器指纹哈希值 """ import platform import uuid # 收集机器信息 机器信息 = { '系统': platform.system(), '节点': platform.node(), '处理器': platform.processor(), 'MAC地址': ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff) for elements in range(0,2*6,2)][::-1]) } # 生成指纹 信息字符串 = json.dumps(机器信息, sort_keys=True) 指纹 = hashlib.sha256(信息字符串.encode()).hexdigest() return 指纹 if __name__ == "__main__": # 测试代码 print("=== JWT工具模块测试 ===") # 测试创建令牌 测试令牌 = 创建访问令牌( 用户ID="test_user", 用户名="测试用户", 有效期天数=7, 角色="管理员", 权限=["交易", "查询", "管理"] ) print(f"生成的令牌: {测试令牌}") # 测试验证令牌 验证结果 = 验证访问令牌(测试令牌) print(f"验证结果: {验证结果}") # 测试获取令牌信息 令牌信息 = 获取令牌信息(测试令牌) print(f"令牌信息: {令牌信息}") # 测试权限检查 权限检查 = 检查令牌权限(测试令牌, "交易") print(f"交易权限检查: {权限检查}") # 测试角色检查 角色检查 = 检查令牌角色(测试令牌, ["管理员", "交易员"]) print(f"角色检查: {角色检查}") # 生成机器指纹 指纹 = 生成机器指纹() print(f"机器指纹: {指纹}")