功能: - StarLight Yggdrasil 第三方登录(后端中转,accessToken 不落地) - 管理员可视化权限管理(用户列表 + 一键授予/移出) - 设置页热配置(11 项可编辑,立即生效) - 循环池(每流独立手动歌单,队列空时随机不重复抽取) - 登录页统一主站浅色风格 - 全站改名「Starlight专属电台」 - 删除关于页 安全: - 全局 CSRF 防护(双提交 Cookie) - 登录限流防 XFF 伪造、错误信息脱敏 - 权限中间件 requireAdminForWrites(读放行/写 admin) - 密码哈希存储(scrypt/sha256) - SSRF 防护(封面下载域名白名单) - 空 catch 加日志、SSE 连接上限 修复: - 多角色账号 selectedProfile 缺失 - 普通用户越权删歌(V4) - streams 写操作越权(V9) - 设置页布局溢出 测试:38 项全绿(api16 / e2e17 / write5)
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
import urllib.request, urllib.error, http.cookiejar, json, sys, os, io, uuid
|
|
|
|
BASE = "http://127.0.0.1:3000"
|
|
import os
|
|
TEST_PASSWORD = os.environ.get("TEST_PASSWORD", "123test")
|
|
cj = http.cookiejar.CookieJar()
|
|
op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
|
|
|
|
def call(path, method="GET", data=None):
|
|
body = json.dumps(data).encode() if data is not None else None
|
|
req = urllib.request.Request(BASE + path, data=body, method=method)
|
|
if body: req.add_header("Content-Type", "application/json")
|
|
if method.upper() not in ("GET","HEAD","OPTIONS"):
|
|
for ck in cj:
|
|
if ck.name == "mr_csrf":
|
|
req.add_header("X-CSRF-Token", ck.value); break
|
|
try:
|
|
r = op.open(req, timeout=30)
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
|
|
results = []
|
|
def check(name, cond, extra=""):
|
|
results.append((name, cond)); print(("PASS" if cond else "FAIL"), name, extra)
|
|
|
|
# 登录
|
|
call("/api/auth/login", "POST", {"password": TEST_PASSWORD})
|
|
|
|
# ---------- 读基线资源数 ----------
|
|
st, body0 = call("/api/resources")
|
|
BASE_N = len(json.loads(body0)["list"])
|
|
print(" 基线资源数:", BASE_N)
|
|
|
|
# ---------- 导入测试 ----------
|
|
tmp_file = r"F:\music_online\demo\tmp\test_import.mp3"
|
|
boundary = "----test" + uuid.uuid4().hex
|
|
with open(tmp_file, "rb") as f:
|
|
audio_data = f.read()
|
|
|
|
title = "测试导入歌"
|
|
artist = "测试作者"
|
|
parts = []
|
|
def add_field(name, value):
|
|
parts.append(("--" + boundary + "\r\n").encode())
|
|
parts.append(('Content-Disposition: form-data; name="%s"\r\n\r\n' % name).encode())
|
|
parts.append((value + "\r\n").encode())
|
|
def add_file(name, filename, content):
|
|
parts.append(("--" + boundary + "\r\n").encode())
|
|
parts.append(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, filename)).encode())
|
|
parts.append(("Content-Type: audio/mpeg\r\n\r\n").encode())
|
|
parts.append(content)
|
|
parts.append(b"\r\n")
|
|
|
|
add_field("title", title)
|
|
add_field("artist", artist)
|
|
add_file("audio", "测试导入歌.mp3", audio_data)
|
|
parts.append(("--" + boundary + "--\r\n").encode())
|
|
body = b"".join(parts)
|
|
|
|
req = urllib.request.Request(BASE + "/api/music/import", data=body, method="POST")
|
|
req.add_header("Content-Type", "multipart/form-data; boundary=" + boundary)
|
|
for ck in cj:
|
|
if ck.name == "mr_csrf":
|
|
req.add_header("X-CSRF-Token", ck.value); break
|
|
try:
|
|
r = op.open(req, timeout=60)
|
|
d = json.loads(r.read())
|
|
check("导入音频", r.status == 200 and d.get("ok"), str(d)[:120])
|
|
imported_file = d.get("file")
|
|
print(" 导入文件名:", imported_file)
|
|
print(" 元数据:", json.dumps(d.get("song", {}), ensure_ascii=False)[:200])
|
|
except urllib.error.HTTPError as e:
|
|
check("导入音频", False, f"HTTP {e.code}: {e.read()[:200]}")
|
|
imported_file = None
|
|
|
|
# 确认资源数增加
|
|
st, body = call("/api/resources")
|
|
d = json.loads(body)
|
|
check(f"资源数变为{BASE_N+1}", len(d["list"]) == BASE_N + 1, f"{len(d['list'])}")
|
|
|
|
# ---------- 删除测试 ----------
|
|
if imported_file:
|
|
st, body = call("/api/music/delete", "POST", {"file": imported_file})
|
|
d = json.loads(body)
|
|
check("删除音频", st == 200 and d.get("ok"), str(d)[:120])
|
|
|
|
# 确认文件没了
|
|
fp = os.path.join(r"F:\music_online\demo\audio", imported_file)
|
|
check("音频文件已删", not os.path.exists(fp))
|
|
# 确认资源数恢复
|
|
st, body = call("/api/resources")
|
|
d = json.loads(body)
|
|
check(f"资源数恢复为{BASE_N}", len(d["list"]) == BASE_N, f"{len(d['list'])}")
|
|
else:
|
|
check("删除音频", False, "无导入文件")
|
|
|
|
print("\n=== 汇总:", sum(1 for _,c in results if c), "/", len(results), "通过 ===")
|
|
sys.exit(0 if all(c for _,c in results) else 1) |