278 lines
7.6 KiB
Python
278 lines
7.6 KiB
Python
#!/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"机器指纹: {指纹}") |