178 lines
5.6 KiB
Python
178 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Redis JWT Token 验证脚本
|
||
用于检查JWT token是否正确存储在Redis中
|
||
|
||
运行方法:
|
||
cd /Users/jiliu/工作/projects/imeeting/backend
|
||
source venv/bin/activate # 激活虚拟环境
|
||
python test/test_redis_jwt.py
|
||
"""
|
||
import sys
|
||
import os
|
||
import redis
|
||
import json
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
try:
|
||
from app.core.config import REDIS_CONFIG
|
||
print("✅ 成功导入项目配置")
|
||
except ImportError as e:
|
||
print(f"❌ 导入项目配置失败: {e}")
|
||
print("请确保在 backend 目录下运行: python test/test_redis_jwt.py")
|
||
sys.exit(1)
|
||
|
||
def check_jwt_in_redis():
|
||
"""检查Redis中的JWT token"""
|
||
try:
|
||
# 使用项目配置连接Redis
|
||
r = redis.Redis(**REDIS_CONFIG)
|
||
|
||
# 测试连接
|
||
r.ping()
|
||
print("✅ Redis连接成功")
|
||
print(f"连接配置: {REDIS_CONFIG}")
|
||
|
||
# 获取所有token相关的keys
|
||
token_keys = r.keys("token:*")
|
||
|
||
if not token_keys:
|
||
print("❌ Redis中没有找到JWT token")
|
||
print("提示: 请先通过前端登录以生成token")
|
||
return False
|
||
|
||
print(f"✅ 找到 {len(token_keys)} 个token记录:")
|
||
|
||
for key in token_keys:
|
||
# 解析key格式: token:user_id:jwt_token
|
||
key_str = key.decode('utf-8') if isinstance(key, bytes) else key
|
||
parts = key_str.split(":", 2)
|
||
if len(parts) >= 3:
|
||
user_id = parts[1]
|
||
token_preview = parts[2][:20] + "..."
|
||
ttl = r.ttl(key)
|
||
value = r.get(key)
|
||
value_str = value.decode('utf-8') if isinstance(value, bytes) else value
|
||
|
||
print(f" - 用户ID: {user_id}")
|
||
print(f" Token预览: {token_preview}")
|
||
if ttl > 0:
|
||
print(f" 剩余时间: {ttl}秒 ({ttl/3600:.1f}小时)")
|
||
else:
|
||
print(f" TTL: {ttl} (永不过期)" if ttl == -1 else f" TTL: {ttl} (已过期)")
|
||
print(f" 状态: {value_str}")
|
||
print()
|
||
|
||
return True
|
||
|
||
except redis.ConnectionError:
|
||
print("❌ 无法连接到Redis服务器")
|
||
print("请确保Redis服务正在运行:")
|
||
print(" brew services start redis # macOS")
|
||
print(" 或 redis-server # 直接启动")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 检查失败: {e}")
|
||
return False
|
||
|
||
def test_token_operations():
|
||
"""测试token操作"""
|
||
try:
|
||
r = redis.Redis(**REDIS_CONFIG)
|
||
|
||
print("\n=== Token操作测试 ===")
|
||
|
||
# 模拟创建token
|
||
test_key = "token:999:test_token_12345"
|
||
r.setex(test_key, 60, "active")
|
||
print(f"✅ 创建测试token: {test_key}")
|
||
|
||
# 检查token存在
|
||
if r.exists(test_key):
|
||
print("✅ Token存在性验证通过")
|
||
|
||
# 检查TTL
|
||
ttl = r.ttl(test_key)
|
||
print(f"✅ Token TTL: {ttl}秒")
|
||
|
||
# 删除测试token
|
||
r.delete(test_key)
|
||
print("✅ 清理测试token")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ Token操作测试失败: {e}")
|
||
return False
|
||
|
||
def test_jwt_service():
|
||
"""测试JWT服务"""
|
||
try:
|
||
from app.services.jwt_service import jwt_service
|
||
|
||
print("\n=== JWT服务测试 ===")
|
||
|
||
# 测试创建token
|
||
test_data = {
|
||
"user_id": 999,
|
||
"username": "test_user",
|
||
"caption": "测试用户"
|
||
}
|
||
|
||
token = jwt_service.create_access_token(test_data)
|
||
print(f"✅ 创建JWT token: {token[:30]}...")
|
||
|
||
# 测试验证token
|
||
payload = jwt_service.verify_token(token)
|
||
if payload:
|
||
print(f"✅ Token验证成功: 用户ID={payload['user_id']}, 用户名={payload['username']}")
|
||
else:
|
||
print("❌ Token验证失败")
|
||
return False
|
||
|
||
# 测试撤销token
|
||
revoked = jwt_service.revoke_token(token, test_data["user_id"])
|
||
print(f"✅ 撤销token: {'成功' if revoked else '失败'}")
|
||
|
||
# 验证撤销后token失效
|
||
payload_after_revoke = jwt_service.verify_token(token)
|
||
if not payload_after_revoke:
|
||
print("✅ Token撤销后验证失败，符合预期")
|
||
else:
|
||
print("❌ Token撤销后仍然有效，不符合预期")
|
||
return False
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ JWT服务测试失败: {e}")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
print("JWT + Redis 认证系统测试")
|
||
print("=" * 50)
|
||
print(f"工作目录: {os.getcwd()}")
|
||
print(f"测试脚本路径: {__file__}")
|
||
|
||
# 检查Redis中的JWT tokens
|
||
redis_ok = check_jwt_in_redis()
|
||
|
||
# 测试token操作
|
||
operations_ok = test_token_operations()
|
||
|
||
# 测试JWT服务
|
||
jwt_service_ok = test_jwt_service()
|
||
|
||
print("=" * 50)
|
||
if redis_ok and operations_ok and jwt_service_ok:
|
||
print("✅ JWT + Redis 认证系统工作正常!")
|
||
else:
|
||
print("❌ JWT + Redis 认证系统存在问题")
|
||
print("\n故障排除建议:")
|
||
print("1. 确保在 backend 目录下运行测试")
|
||
print("2. 确保Redis服务正在运行")
|
||
print("3. 确保已安装所有依赖: pip install -r requirements.txt")
|
||
print("4. 尝试先通过前端登录生成token")
|
||
sys.exit(1) |