483 lines
14 KiB
Python
483 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
CarrotSkin 测试账户生成器
|
||
此脚本创建新用户、角色，并输出所有测试信息
|
||
"""
|
||
|
||
import json
|
||
import random
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
try:
|
||
import requests
|
||
except ImportError:
|
||
print("错误: 需要安装 requests 库")
|
||
print("请运行: pip install requests")
|
||
sys.exit(1)
|
||
|
||
# 颜色输出支持（Windows 和 Unix 都支持）
|
||
try:
|
||
from colorama import init, Fore, Style
|
||
init(autoreset=True)
|
||
HAS_COLORAMA = True
|
||
# colorama 没有 GRAY，使用 WHITE 代替
|
||
Fore.GRAY = Fore.WHITE
|
||
except ImportError:
|
||
# 如果没有 colorama，使用空字符串
|
||
class Fore:
|
||
CYAN = YELLOW = GREEN = RED = GRAY = WHITE = RESET = ""
|
||
class Style:
|
||
RESET_ALL = ""
|
||
HAS_COLORAMA = False
|
||
|
||
# API 基础 URL
|
||
BASE_URL = "http://localhost:8080/api/v1"
|
||
|
||
|
||
def print_colored(text, color=Fore.RESET):
|
||
"""打印彩色文本"""
|
||
print(f"{color}{text}{Style.RESET_ALL}")
|
||
|
||
|
||
def print_header(text):
|
||
"""打印标题"""
|
||
print_colored(f"\n{'=' * 40}", Fore.CYAN)
|
||
print_colored(f" {text}", Fore.CYAN)
|
||
print_colored(f"{'=' * 40}\n", Fore.CYAN)
|
||
|
||
|
||
def print_step(text):
|
||
"""打印步骤标题"""
|
||
print_colored(f"\n=== {text} ===", Fore.YELLOW)
|
||
|
||
|
||
def print_success(text):
|
||
"""打印成功消息"""
|
||
print_colored(f"✓ {text}", Fore.GREEN)
|
||
|
||
|
||
def print_error(text):
|
||
"""打印错误消息"""
|
||
print_colored(f"✗ {text}", Fore.RED)
|
||
|
||
|
||
def print_info(text):
|
||
"""打印信息消息"""
|
||
print_colored(f" {text}", Fore.GRAY)
|
||
|
||
|
||
def register_user():
|
||
"""步骤1: 注册新用户"""
|
||
print_step("步骤1: 注册新用户")
|
||
|
||
random_num = random.randint(10000, 99999)
|
||
username = f"testuser{random_num}"
|
||
email = f"test{random_num}@example.com"
|
||
login_password = "password123"
|
||
verification_code = "123456"
|
||
|
||
print_info(f"用户名: {username}")
|
||
print_info(f"邮箱: {email}")
|
||
print_info(f"密码: {login_password}")
|
||
|
||
register_data = {
|
||
"username": username,
|
||
"email": email,
|
||
"password": login_password,
|
||
"verification_code": verification_code
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/auth/register",
|
||
json=register_data,
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
jwt_token = result["data"]["token"]
|
||
user_id = result["data"]["user_info"]["id"]
|
||
user_role = result["data"]["user_info"]["role"]
|
||
|
||
print_success("注册成功!")
|
||
print_info(f"用户ID: {user_id}")
|
||
print_info(f"角色: {user_role}")
|
||
|
||
return {
|
||
"username": username,
|
||
"email": email,
|
||
"login_password": login_password,
|
||
"jwt_token": jwt_token,
|
||
"user_id": user_id,
|
||
"user_role": user_role
|
||
}
|
||
else:
|
||
print_error(f"注册失败: {result.get('message', '未知错误')}")
|
||
sys.exit(1)
|
||
except requests.exceptions.RequestException as e:
|
||
print_error(f"注册失败: {str(e)}")
|
||
sys.exit(1)
|
||
|
||
|
||
def create_profile(jwt_token):
|
||
"""步骤2: 创建Minecraft角色"""
|
||
print_step("步骤2: 创建Minecraft角色")
|
||
|
||
profile_name = f"TestPlayer{random.randint(100, 999)}"
|
||
print_info(f"角色名: {profile_name}")
|
||
|
||
profile_data = {
|
||
"name": profile_name
|
||
}
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {jwt_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/profile/",
|
||
json=profile_data,
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
profile_uuid = result["data"]["uuid"]
|
||
print_success("角色创建成功!")
|
||
print_info(f"角色UUID: {profile_uuid}")
|
||
print_info(f"角色名: {result['data']['name']}")
|
||
|
||
return {
|
||
"profile_name": profile_name,
|
||
"profile_uuid": profile_uuid
|
||
}
|
||
else:
|
||
print_error(f"角色创建失败: {result.get('message', '未知错误')}")
|
||
sys.exit(1)
|
||
except requests.exceptions.RequestException as e:
|
||
print_error(f"角色创建失败: {str(e)}")
|
||
sys.exit(1)
|
||
|
||
|
||
def reset_yggdrasil_password(jwt_token):
|
||
"""步骤3: 重置并获取Yggdrasil密码"""
|
||
print_step("步骤3: 重置并获取Yggdrasil密码")
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {jwt_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/user/yggdrasil-password/reset",
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
yggdrasil_password = result["data"]["password"]
|
||
print_success("Yggdrasil密码重置成功!")
|
||
print_info(f"Yggdrasil密码: {yggdrasil_password}")
|
||
return yggdrasil_password
|
||
else:
|
||
print_error(f"Yggdrasil密码重置失败: {result.get('message', '未知错误')}")
|
||
return ""
|
||
except requests.exceptions.RequestException as e:
|
||
print_error(f"Yggdrasil密码重置失败: {str(e)}")
|
||
return ""
|
||
|
||
|
||
def generate_output(user_info, profile_info, yggdrasil_password):
|
||
"""生成输出信息"""
|
||
output = f"""========================================
|
||
CarrotSkin 测试账户信息
|
||
========================================
|
||
|
||
=== 账户信息 ===
|
||
用户名: {user_info['username']}
|
||
邮箱: {user_info['email']}
|
||
登录密码: {user_info['login_password']}
|
||
用户ID: {user_info['user_id']}
|
||
|
||
=== JWT Token (API认证) ===
|
||
Token: {user_info['jwt_token']}
|
||
|
||
=== 角色信息 ===
|
||
角色名: {profile_info['profile_name']}
|
||
角色UUID: {profile_info['profile_uuid']}
|
||
|
||
=== Yggdrasil信息 ===
|
||
Yggdrasil密码: {yggdrasil_password}
|
||
|
||
=== 测试命令 ===
|
||
|
||
# 1. API登录
|
||
curl -X POST http://localhost:8080/api/v1/auth/login \\
|
||
-H 'Content-Type: application/json' \\
|
||
-d '{{"username":"{user_info['username']}","password":"{user_info['login_password']}"}}'
|
||
|
||
# 2. 创建角色
|
||
curl -X POST http://localhost:8080/api/v1/profile/ \\
|
||
-H 'Content-Type: application/json' \\
|
||
-H 'Authorization: Bearer {user_info['jwt_token']}' \\
|
||
-d '{{"name":"NewProfile"}}'
|
||
|
||
# 3. 重置Yggdrasil密码
|
||
curl -X POST http://localhost:8080/api/v1/user/yggdrasil-password/reset \\
|
||
-H 'Content-Type: application/json' \\
|
||
-H 'Authorization: Bearer {user_info['jwt_token']}'
|
||
|
||
# 4. Yggdrasil认证
|
||
curl -X POST http://localhost:8080/api/v1/yggdrasil/authserver/authenticate \\
|
||
-H 'Content-Type: application/json' \\
|
||
-d '{{
|
||
"username": "{user_info['username']}",
|
||
"password": "{yggdrasil_password}",
|
||
"requestUser": true
|
||
}}'
|
||
|
||
# 5. 获取角色信息
|
||
curl -X GET http://localhost:8080/api/v1/profile/{profile_info['profile_uuid']}
|
||
|
||
========================================
|
||
"""
|
||
return output
|
||
|
||
|
||
def set_user_role(admin_token, user_id, role):
|
||
"""设置用户角色"""
|
||
headers = {
|
||
"Authorization": f"Bearer {admin_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
try:
|
||
response = requests.put(
|
||
f"{BASE_URL}/admin/users/role",
|
||
json={"user_id": user_id, "role": role},
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
print_success(f"用户角色设置为: {role}")
|
||
return True
|
||
else:
|
||
print_error(f"设置角色失败: {result.get('message', '未知错误')}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print_error(f"设置角色失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
def login_user(username, password):
|
||
"""登录用户"""
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/auth/login",
|
||
json={"username": username, "password": password},
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
return result["data"]["token"]
|
||
return None
|
||
except requests.exceptions.RequestException:
|
||
return None
|
||
|
||
|
||
def create_admin_user():
|
||
"""创建管理员用户"""
|
||
print_step("创建管理员用户")
|
||
|
||
random_num = random.randint(10000, 99999)
|
||
username = f"admin{random_num}"
|
||
email = f"admin{random_num}@example.com"
|
||
login_password = "admin123456"
|
||
verification_code = "123456"
|
||
|
||
print_info(f"用户名: {username}")
|
||
print_info(f"邮箱: {email}")
|
||
print_info(f"密码: {login_password}")
|
||
|
||
register_data = {
|
||
"username": username,
|
||
"email": email,
|
||
"password": login_password,
|
||
"verification_code": verification_code
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/auth/register",
|
||
json=register_data,
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
jwt_token = result["data"]["token"]
|
||
user_id = result["data"]["user_info"]["id"]
|
||
|
||
print_success("用户注册成功!")
|
||
print_info(f"用户ID: {user_id}")
|
||
|
||
# 使用默认管理员账户提升权限
|
||
print_info("尝试使用默认管理员账户提升权限...")
|
||
default_admin_token = login_user("admin", "admin123456")
|
||
|
||
if default_admin_token:
|
||
if set_user_role(default_admin_token, user_id, "admin"):
|
||
print_success("管理员权限设置成功!")
|
||
return {
|
||
"username": username,
|
||
"email": email,
|
||
"password": login_password,
|
||
"jwt_token": jwt_token,
|
||
"user_id": user_id,
|
||
"role": "admin"
|
||
}
|
||
|
||
print_error("无法提升权限，请手动设置")
|
||
return {
|
||
"username": username,
|
||
"email": email,
|
||
"password": login_password,
|
||
"jwt_token": jwt_token,
|
||
"user_id": user_id,
|
||
"role": "user"
|
||
}
|
||
else:
|
||
print_error(f"注册失败: {result.get('message', '未知错误')}")
|
||
return None
|
||
except requests.exceptions.RequestException as e:
|
||
print_error(f"注册失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
def generate_admin_output(admin_info):
|
||
"""生成管理员账户输出信息"""
|
||
output = f"""========================================
|
||
CarrotSkin 管理员账户信息
|
||
========================================
|
||
|
||
=== 账户信息 ===
|
||
用户名: {admin_info['username']}
|
||
邮箱: {admin_info['email']}
|
||
密码: {admin_info['password']}
|
||
用户ID: {admin_info['user_id']}
|
||
角色: {admin_info['role']}
|
||
|
||
=== JWT Token (API认证) ===
|
||
Token: {admin_info['jwt_token']}
|
||
|
||
=== 管理员API示例 ===
|
||
|
||
# 1. 获取用户列表
|
||
curl -X GET "{BASE_URL}/admin/users" \\
|
||
-H "Authorization: Bearer {admin_info['jwt_token']}"
|
||
|
||
# 2. 设置用户角色为管理员
|
||
curl -X PUT "{BASE_URL}/admin/users/role" \\
|
||
-H "Content-Type: application/json" \\
|
||
-H "Authorization: Bearer {admin_info['jwt_token']}" \\
|
||
-d '{{"user_id": 1, "role": "admin"}}'
|
||
|
||
# 3. 获取材质列表（审核）
|
||
curl -X GET "{BASE_URL}/admin/textures" \\
|
||
-H "Authorization: Bearer {admin_info['jwt_token']}"
|
||
|
||
# 4. 删除材质
|
||
curl -X DELETE "{BASE_URL}/admin/textures/1" \\
|
||
-H "Authorization: Bearer {admin_info['jwt_token']}"
|
||
|
||
========================================
|
||
"""
|
||
return output
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print_header("CarrotSkin 测试账户生成器")
|
||
|
||
# 选择模式
|
||
print("请选择操作:")
|
||
print(" 1. 创建普通测试用户")
|
||
print(" 2. 创建管理员用户")
|
||
print(" 3. 创建两者")
|
||
|
||
choice = input("\n请输入选项 (1/2/3) [默认: 1]: ").strip() or "1"
|
||
|
||
if choice in ["1", "3"]:
|
||
# 创建普通用户
|
||
print_header("创建普通测试用户")
|
||
|
||
# 步骤1: 注册用户
|
||
user_info = register_user()
|
||
|
||
# 步骤2: 创建角色
|
||
profile_info = create_profile(user_info["jwt_token"])
|
||
|
||
# 步骤3: 重置Yggdrasil密码
|
||
yggdrasil_password = reset_yggdrasil_password(user_info["jwt_token"])
|
||
|
||
# 输出信息
|
||
print_header("普通用户信息汇总")
|
||
output = generate_output(user_info, profile_info, yggdrasil_password)
|
||
print(output)
|
||
|
||
# 保存到文件
|
||
output_file = f"test_account_{user_info['username']}.txt"
|
||
try:
|
||
with open(output_file, "w", encoding="utf-8") as f:
|
||
f.write(output)
|
||
print_success(f"信息已保存到文件: {output_file}")
|
||
except Exception as e:
|
||
print_error(f"保存文件失败: {str(e)}")
|
||
|
||
if choice in ["2", "3"]:
|
||
# 创建管理员用户
|
||
print_header("创建管理员用户")
|
||
admin_info = create_admin_user()
|
||
|
||
if admin_info:
|
||
print_header("管理员账户信息汇总")
|
||
admin_output = generate_admin_output(admin_info)
|
||
print(admin_output)
|
||
|
||
# 保存到文件
|
||
admin_output_file = f"admin_account_{admin_info['username']}.txt"
|
||
try:
|
||
with open(admin_output_file, "w", encoding="utf-8") as f:
|
||
f.write(admin_output)
|
||
print_success(f"管理员信息已保存到文件: {admin_output_file}")
|
||
except Exception as e:
|
||
print_error(f"保存文件失败: {str(e)}")
|
||
|
||
print_header("测试完成!")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|