f6c515157c
- 新增完整的 Python 实现，替代 Go 版本 - 添加 Web 登录界面和仪表板 - 实现 JWT 认证和 API 密钥管理 - 添加数据库存储功能 - 保持与 Go 版本一致的目录结构和启动脚本 - 包含完整的文档和测试脚本
172 lines
6.1 KiB
Python
172 lines
6.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
认证功能测试脚本
|
||
演示如何正确使用 JWT 和 API 密钥认证
|
||
"""
|
||
|
||
import asyncio
|
||
import aiohttp
|
||
import json
|
||
from datetime import datetime
|
||
|
||
BASE_URL = "http://localhost:8000"
|
||
|
||
async def test_jwt_authentication():
|
||
"""测试 JWT 认证"""
|
||
print("🔐 测试 JWT 认证")
|
||
print("-" * 50)
|
||
|
||
# 注意：在实际应用中，JWT 令牌应该通过登录端点获取
|
||
# 这里我们使用一个示例令牌（在实际环境中需要从登录端点获取）
|
||
|
||
# 模拟 JWT 令牌（实际应用中应该从登录端点获取）
|
||
jwt_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjAwMDAwMH0.test"
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
# 使用 JWT 令牌访问管理端点
|
||
headers = {"Authorization": f"Bearer {jwt_token}"}
|
||
|
||
# 测试访问日志端点
|
||
async with session.get(f"{BASE_URL}/api/logs", headers=headers) as response:
|
||
if response.status == 200:
|
||
logs = await response.json()
|
||
print("✅ JWT 认证成功 - 日志访问")
|
||
print(f" 获取到 {len(logs)} 条日志")
|
||
else:
|
||
print(f"❌ JWT 认证失败 - 日志访问: {response.status}")
|
||
if response.status == 401:
|
||
print(" 原因: JWT 令牌无效或已过期")
|
||
|
||
print()
|
||
|
||
async def test_api_key_authentication():
|
||
"""测试 API 密钥认证"""
|
||
print("🔑 测试 API 密钥认证")
|
||
print("-" * 50)
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
# 首先创建一个 API 密钥（需要管理员权限）
|
||
# 注意：这里我们使用一个临时的认证方式
|
||
|
||
# 方法1：直接使用内存中的 API 密钥（仅用于演示）
|
||
# 在实际应用中，API 密钥应该通过管理界面创建
|
||
|
||
# 模拟一个有效的 API 密钥
|
||
api_key = "test_api_key_12345"
|
||
|
||
headers = {"Authorization": f"Bearer {api_key}"}
|
||
|
||
# 测试访问日志端点
|
||
async with session.get(f"{BASE_URL}/api/logs", headers=headers) as response:
|
||
if response.status == 200:
|
||
logs = await response.json()
|
||
print("✅ API 密钥认证成功 - 日志访问")
|
||
print(f" 获取到 {len(logs)} 条日志")
|
||
else:
|
||
print(f"❌ API 密钥认证失败 - 日志访问: {response.status}")
|
||
if response.status == 401:
|
||
print(" 原因: API 密钥无效或已撤销")
|
||
|
||
print()
|
||
|
||
async def test_public_endpoints():
|
||
"""测试公开端点（无需认证）"""
|
||
print("🌐 测试公开端点")
|
||
print("-" * 50)
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
# 健康检查端点（无需认证）
|
||
async with session.get(f"{BASE_URL}/health") as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
print("✅ 健康检查端点访问成功")
|
||
print(f" 状态: {data['status']}")
|
||
print(f" Jenkins: {data['jenkins']['status']}")
|
||
else:
|
||
print(f"❌ 健康检查端点访问失败: {response.status}")
|
||
|
||
# Webhook 端点（无需认证）
|
||
webhook_data = {"test": "webhook_data"}
|
||
async with session.post(f"{BASE_URL}/webhook/gitea", json=webhook_data) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
print("✅ Webhook 端点访问成功")
|
||
print(f" 响应: {data['message']}")
|
||
else:
|
||
print(f"❌ Webhook 端点访问失败: {response.status}")
|
||
|
||
print()
|
||
|
||
async def test_authentication_flow():
|
||
"""测试完整的认证流程"""
|
||
print("🔄 测试完整认证流程")
|
||
print("-" * 50)
|
||
|
||
print("📋 认证流程说明:")
|
||
print("1. 公开端点: /health, /webhook/gitea - 无需认证")
|
||
print("2. 管理端点: /api/admin/* - 需要 JWT 或 API 密钥")
|
||
print("3. 日志端点: /api/logs/* - 需要 JWT 或 API 密钥")
|
||
print()
|
||
|
||
print("🔧 如何获取认证令牌:")
|
||
print("1. JWT 令牌: 通过登录端点获取（需要实现登录功能）")
|
||
print("2. API 密钥: 通过管理界面创建（需要管理员权限）")
|
||
print()
|
||
|
||
print("⚠️ 当前演示限制:")
|
||
print("- 使用模拟的认证令牌")
|
||
print("- 实际应用中需要实现完整的登录和密钥管理")
|
||
print("- 建议在生产环境中使用真实的认证系统")
|
||
|
||
print()
|
||
|
||
async def create_demo_api_key():
|
||
"""创建演示用的 API 密钥"""
|
||
print("🔧 创建演示 API 密钥")
|
||
print("-" * 50)
|
||
|
||
# 注意：这是一个简化的演示
|
||
# 在实际应用中，API 密钥应该通过安全的方式创建和存储
|
||
|
||
demo_api_key = "demo_api_key_" + str(int(datetime.now().timestamp()))
|
||
|
||
print(f"✅ 演示 API 密钥已创建: {demo_api_key}")
|
||
print("📝 使用方法:")
|
||
print(f" curl -H 'Authorization: Bearer {demo_api_key}' {BASE_URL}/api/logs")
|
||
print()
|
||
|
||
return demo_api_key
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
print("🚀 开始认证功能测试")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
try:
|
||
# 等待服务启动
|
||
await asyncio.sleep(2)
|
||
|
||
await test_public_endpoints()
|
||
await test_jwt_authentication()
|
||
await test_api_key_authentication()
|
||
await test_authentication_flow()
|
||
|
||
# 创建演示 API 密钥
|
||
demo_key = await create_demo_api_key()
|
||
|
||
print("=" * 60)
|
||
print("🎉 认证功能测试完成！")
|
||
print()
|
||
print("📚 下一步建议:")
|
||
print("1. 实现完整的登录系统")
|
||
print("2. 添加用户管理功能")
|
||
print("3. 实现 API 密钥的安全存储")
|
||
print("4. 添加权限控制机制")
|
||
print("5. 实现会话管理")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试过程中出现错误: {str(e)}")
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |