zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
323 lines
16 KiB
Python
323 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""纯逻辑自测（不启动任何网络服务）：DER/RSA、可信地址闸门、下载登记去重、配置校验。
|
||
|
||
python3 -m unittest discover -s tests -v
|
||
或 python3 tests/test_basic.py
|
||
"""
|
||
import contextlib
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from ktv import musesync, resolver
|
||
from ktv.config import Config
|
||
from ktv.store import Store
|
||
|
||
|
||
class TestRsa(unittest.TestCase):
|
||
def test_der_extracts_rsa_numbers(self):
|
||
n, e = musesync.rsa_public_numbers()
|
||
self.assertEqual(e, 65537)
|
||
# 2048 位密钥
|
||
self.assertTrue(2 ** 2047 < n < 2 ** 2048)
|
||
|
||
def test_pkcs1v15_layout(self):
|
||
n, e = musesync.rsa_public_numbers()
|
||
k = (n.bit_length() + 7) // 8
|
||
msg = b'{"device_id":"s00e04b021b2d"}'
|
||
ct = musesync.rsa_pkcs1v15_encrypt(msg, n, e)
|
||
self.assertEqual(len(ct), k)
|
||
# 无法解密（没有私钥），但可验证密文长度与 base64 往返
|
||
import base64
|
||
blob = musesync.encrypt_device_id("s00e04b021b2d")
|
||
self.assertEqual(len(base64.b64decode(blob)), k)
|
||
|
||
def test_random_padding_differs(self):
|
||
n, e = musesync.rsa_public_numbers()
|
||
a = musesync.rsa_pkcs1v15_encrypt(b"x", n, e)
|
||
b = musesync.rsa_pkcs1v15_encrypt(b"x", n, e)
|
||
self.assertNotEqual(a, b) # PKCS#1 v1.5 随机填充
|
||
|
||
|
||
class TestGate(unittest.TestCase):
|
||
"""可信地址闸门：必须与 App 的 isDemoUrl / isValidSongUrl 一致。"""
|
||
|
||
def test_accepts_real_cdn(self):
|
||
ok, _ = resolver.is_trusted_url(
|
||
"http://txog.ktvsky.com/264/lvbr/480/1735508576056/E/ts/83.1/202408120244/480p/7009126.ts?token=x", 7009126)
|
||
self.assertTrue(ok)
|
||
|
||
def test_rejects_bare_ip(self):
|
||
ok, why = resolver.is_trusted_url("http://1.2.3.4/media/7009126.ts", 7009126)
|
||
self.assertFalse(ok)
|
||
self.assertIn("裸 IP", why)
|
||
|
||
def test_rejects_wrong_filename(self):
|
||
ok, _ = resolver.is_trusted_url("http://cdn.example.com/x/719.ts", 7009126)
|
||
self.assertFalse(ok)
|
||
|
||
def test_rejects_ad_hosts_and_markers(self):
|
||
for url in ("http://gz.ac16.vip/ad_files/my_ad_video.ts",
|
||
"https://wb66.cn/f/d/7b7c9/10188555.ts",
|
||
"https://pub-abc123.r2.dev/ss/demo.ts"):
|
||
ok, _ = resolver.is_trusted_url(url, 7009126)
|
||
self.assertFalse(ok, url)
|
||
|
||
def test_accepts_own_domain(self):
|
||
ok, _ = resolver.is_trusted_url("http://ktv.example.com:6677/media/7009126.ts", 7009126)
|
||
self.assertTrue(ok)
|
||
|
||
|
||
class TestQueryStringGate(unittest.TestCase):
|
||
"""⚠️ 带 query 的媒体地址（2026-10-11 抓包形态）必须判为合法。
|
||
|
||
真源地址恒为 `…/<编号>.ts?token=…&t=…`（`txog.ktvsky.com` / `download.origjoy.com`）。
|
||
这里锁住三件事：
|
||
1. `.ts` 文件名比对**先剥离 query/fragment**（否则要么误杀真源、要么漏放广告）；
|
||
2. 广告黑名单（wb66/demo/ac16/ad_files/r2.dev）**在带 query 时依然生效**；
|
||
3. query 里的随机串（如 `note=demo`）**不再误杀**合法地址 —— 只在 host+path 上判特征。
|
||
"""
|
||
|
||
HAR_URL = ("http://txog.ktvsky.com/264/lvbr/480/E/ts/39.6/202309070206/480p/"
|
||
"7200088.ts?token=432fe2b3345588df379f5e2da3100617&t=1791706901")
|
||
|
||
def test_accepts_exact_har_url(self):
|
||
"""D：新返回域名 txog.ktvsky.com + query 串 → 必须通过（我们没有域名白名单，只有黑名单）。"""
|
||
ok, why = resolver.is_trusted_url(self.HAR_URL, 7200088)
|
||
self.assertTrue(ok, why)
|
||
self.assertEqual(resolver.ts_basename(self.HAR_URL), "7200088")
|
||
|
||
def test_accepts_origjoy_url_with_query(self):
|
||
ok, _ = resolver.is_trusted_url(
|
||
"http://download.origjoy.com/264/lvbr/480/E/ts/39.6/202309070206/480p/7445418.ts"
|
||
"?token=abc&t=1791706901", 7445418)
|
||
self.assertTrue(ok)
|
||
|
||
def test_filename_check_works_with_query_and_fragment(self):
|
||
"""带 query/fragment 时文件名不符**必须**被拦（旧实现在 `#` 上是漏放的）。"""
|
||
for url in ("http://cdn.example.com/x/9999999.ts?token=x",
|
||
"http://cdn.example.com/x/9999999.ts#frag",
|
||
"http://cdn.example.com/x/9999999.ts?token=x&t=1#frag"):
|
||
ok, why = resolver.is_trusted_url(url, 7200088)
|
||
self.assertFalse(ok, url)
|
||
self.assertIn("9999999", why)
|
||
|
||
def test_matching_name_with_query_or_fragment_is_ok(self):
|
||
for url in ("http://cdn.example.com/x/7200088.ts?token=x&t=1",
|
||
"http://cdn.example.com/x/7200088.ts#frag",
|
||
"http://cdn.example.com/x/7200088.ts?a=1#frag"):
|
||
ok, why = resolver.is_trusted_url(url, 7200088)
|
||
self.assertTrue(ok, f"{url} {why}")
|
||
|
||
def test_ad_targets_still_rejected_with_query(self):
|
||
"""C：带 query 时黑名单依旧生效（wb66 / demo / ac16 / ad_files / r2.dev）。"""
|
||
for url, tid in (("https://wb66.cn/f/d/7b7c9/10188555.ts?token=x", 7200088),
|
||
("https://wb66.cn/f/d/7b7c9/7200088.ts?token=x", 7200088),
|
||
("http://demo.example.com/x/7200088.ts?token=x", 7200088),
|
||
("http://gz.ac16.vip/ad_files/my_ad_video.ts?token=x", 7200088),
|
||
("http://cdn.ac16.vip/x/7200088.ts?token=x", 7200088),
|
||
("https://pub-abc123.r2.dev/ss/7200088.ts?token=x", 7200088)):
|
||
ok, why = resolver.is_trusted_url(url, tid)
|
||
self.assertFalse(ok, f"{url} → {why}")
|
||
|
||
def test_query_value_with_demo_word_is_not_a_false_reject(self):
|
||
"""query 里的随机串（签名/参数）不参与广告特征匹配 —— 只判 host+path。"""
|
||
ok, why = resolver.is_trusted_url(
|
||
"http://txog.ktvsky.com/264/lvbr/480/E/ts/39.6/x/480p/7200088.ts?note=demo&token=wb66",
|
||
7200088)
|
||
self.assertTrue(ok, why)
|
||
|
||
|
||
class TestCanonicalSign(unittest.TestCase):
|
||
"""A：用**现实现的拼串**复算 2026-10-11 抓包的 sign（逐字节相等 = 取歌参数没变）。
|
||
|
||
抓包（Reqable，`zk.ktv1.top:6677`）：
|
||
* `GET /i.php?appid=…&mac=ae60e9547958_2d7bb4a7d7f8&sn=2d7bb4a7d7f8&time=1791702376
|
||
&ver=2.0&vn=4.1.3.03281430&sign=5ff87db502fb1d4f5336a190210722d6`
|
||
* `GET /music/do.php?appid=…&device=ae60e9547958_2d7bb4a7d7f8&ish265=0&ls=1&musicno=7200088
|
||
&resolution=720&sn=2d7bb4a7d7f8&time=1791702377&token=NjI1…&sign=4522dcbe72c0f583029f6f1b6fa7850b`
|
||
"""
|
||
|
||
HAR_APPID = "d4eeacc6cec3434fbc8c41608a3056a0"
|
||
HAR_MAC = "ae60e9547958"
|
||
HAR_SN = "2d7bb4a7d7f8"
|
||
HAR_TOKEN = "NjI1YjUyY2ZmYzExYjhjNDI2MjVkMDQwMzI4MmJjZGU="
|
||
HAR_I_TIME = 1791702376
|
||
HAR_D_TIME = 1791702377
|
||
HAR_I_SIGN = "5ff87db502fb1d4f5336a190210722d6"
|
||
HAR_D_SIGN = "4522dcbe72c0f583029f6f1b6fa7850b"
|
||
HAR_I_PARAMS = ("appid=d4eeacc6cec3434fbc8c41608a3056a0&mac=ae60e9547958_2d7bb4a7d7f8"
|
||
"&sn=2d7bb4a7d7f8&time=1791702376&ver=2.0&vn=4.1.3.03281430")
|
||
HAR_D_PARAMS = ("appid=d4eeacc6cec3434fbc8c41608a3056a0&device=ae60e9547958_2d7bb4a7d7f8"
|
||
"&ish265=0&ls=1&musicno=7200088&resolution=720&sn=2d7bb4a7d7f8"
|
||
"&time=1791702377&token=NjI1YjUyY2ZmYzExYjhjNDI2MjVkMDQwMzI4MmJjZGU=")
|
||
HAR_I_URL = "http://zk.ktv1.top:6677/i.php?" + "appid=d4eeacc6cec3434fbc8c41608a3056a0&mac=ae60e9547958_2d7bb4a7d7f8&sn=2d7bb4a7d7f8&time=1791702376&ver=2.0&vn=4.1.3.03281430&sign=5ff87db502fb1d4f5336a190210722d6"
|
||
HAR_D_URL = ("http://zk.ktv1.top:6677/music/do.php?appid=d4eeacc6cec3434fbc8c41608a3056a0"
|
||
"&device=ae60e9547958_2d7bb4a7d7f8&ish265=0&ls=1&musicno=7200088&resolution=720"
|
||
"&sn=2d7bb4a7d7f8&time=1791702377"
|
||
"&token=NjI1YjUyY2ZmYzExYjhjNDI2MjVkMDQwMzI4MmJjZGU="
|
||
"&sign=4522dcbe72c0f583029f6f1b6fa7850b")
|
||
|
||
def test_keys_unchanged(self):
|
||
self.assertEqual(resolver.APP_ID, self.HAR_APPID)
|
||
self.assertEqual(resolver.APP_KEY, "024210cba40d4385a93e6c2d3249bfb5")
|
||
self.assertEqual(resolver.SDK_KEY, "19042303a8374f67ae3fe1e25c97936f")
|
||
self.assertEqual(resolver.VER, "2.0")
|
||
self.assertEqual(resolver.VN, "4.1.3.03281430")
|
||
|
||
def test_token_params_and_sign_match_har_byte_for_byte(self):
|
||
params = resolver.token_params(self.HAR_MAC, self.HAR_SN, ts=self.HAR_I_TIME)
|
||
self.assertEqual(params, self.HAR_I_PARAMS)
|
||
self.assertEqual(resolver.token_sign(params), self.HAR_I_SIGN)
|
||
self.assertEqual(resolver.md5(params + resolver.APP_KEY), self.HAR_I_SIGN)
|
||
|
||
def test_song_params_and_sign_match_har_byte_for_byte(self):
|
||
params = resolver.song_params(7200088, self.HAR_TOKEN, self.HAR_MAC, self.HAR_SN,
|
||
"720", False, ts=self.HAR_D_TIME)
|
||
self.assertEqual(params, self.HAR_D_PARAMS)
|
||
self.assertEqual(resolver.song_sign(params), self.HAR_D_SIGN)
|
||
# token 里的 `=` 必须原样参与签名（不能再编码）
|
||
self.assertIn("token=" + self.HAR_TOKEN, params)
|
||
|
||
def test_live_get_token_builds_exact_har_url(self):
|
||
"""**现网代码路径**（get_token）在抓包时间戳下必须产出与 HAR 完全一致的 URL。"""
|
||
seen: dict = {}
|
||
with self._capture(seen), mock.patch.object(resolver.time, "time", return_value=self.HAR_I_TIME):
|
||
token = resolver.get_token("http://zk.ktv1.top:6677", self.HAR_MAC, self.HAR_SN)
|
||
self.assertEqual(token, self.HAR_TOKEN)
|
||
self.assertEqual(seen["url"], self.HAR_I_URL)
|
||
|
||
def test_live_get_song_url_builds_exact_har_url(self):
|
||
seen: dict = {}
|
||
with self._capture(seen), mock.patch.object(resolver.time, "time", return_value=self.HAR_D_TIME):
|
||
url, detail = resolver.get_song_url("http://zk.ktv1.top:6677", 7200088, self.HAR_TOKEN,
|
||
self.HAR_MAC, self.HAR_SN, "720")
|
||
self.assertEqual(seen["url"], self.HAR_D_URL)
|
||
self.assertEqual(detail, "ok")
|
||
self.assertEqual(url, "http://txog.ktvsky.com/264/lvbr/480/E/ts/39.6/202309070206/"
|
||
"480p/7200088.ts?token=432fe2b3345588df379f5e2da3100617&t=1791706901")
|
||
|
||
# ── 辅助 ──
|
||
@contextlib.contextmanager
|
||
def _capture(self, seen: dict):
|
||
"""把 `_get` 换成"记录 URL 并返回抓包响应"的假实现（不联网、不限速）。"""
|
||
def fake_get(url, *args, **kwargs):
|
||
seen["url"] = url
|
||
if "/i.php" in url:
|
||
return 200, ('{"code":200,"end_date":"2026-10-05 14:58:21","ordertype":0,'
|
||
'"online_state":0,"token":"NjI1YjUyY2ZmYzExYjhjNDI2MjVkMDQwMzI4MmJjZGU=",'
|
||
'"retry":5,"msg":"ok","heart_time":5400}')
|
||
return 200, ('{"msg":"SUCCESS","data":"http://txog.ktvsky.com/264/lvbr/480/E/ts/39.6/'
|
||
'202309070206/480p/7200088.ts?token=432fe2b3345588df379f5e2da3100617'
|
||
'&t=1791706901","code":200,"limit":2000}')
|
||
with mock.patch.object(resolver, "_get", fake_get):
|
||
yield
|
||
|
||
|
||
class TestStoreDedupe(unittest.TestCase):
|
||
"""下载登记与去重：这是"避免重复下载"的核心。"""
|
||
|
||
def tearDown(self):
|
||
# 显式关连接：否则 ResourceWarning（unclosed database）会污染验收输出
|
||
try:
|
||
self.store.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def setUp(self):
|
||
self.tmp = tempfile.mkdtemp()
|
||
self.store = Store(os.path.join(self.tmp, "t.db"))
|
||
rows = [(7009126, "一帘好梦", "国语", "", "胡云鹏", "7009126.ts", 1060, ""),
|
||
(7303235, "测试歌", "粤语", "", "某人", "7303235.ts", 500, "")]
|
||
self.store.import_library(rows, "test-version")
|
||
|
||
def test_claim_is_atomic(self):
|
||
self.store.enqueue([7009126])
|
||
self.assertTrue(self.store.claim_one(7009126, backoff_sec=0)) # 第一次抢到
|
||
self.assertFalse(self.store.claim_one(7009126, backoff_sec=0)) # 第二次抢不到（已在下载）
|
||
self.store.finish(7009126, True, bytes_=123, path="/tmp/x.ts")
|
||
self.assertFalse(self.store.claim_one(7009126, backoff_sec=0)) # 已完成，永不重复下载
|
||
|
||
def test_enqueue_skips_done(self):
|
||
self.store.enqueue([7009126])
|
||
self.store.claim_one(7009126, backoff_sec=0)
|
||
self.store.finish(7009126, True, bytes_=999)
|
||
res = self.store.enqueue([7009126])
|
||
self.assertEqual(res["skipped_done"], 1)
|
||
self.assertEqual(res["added"], 0)
|
||
|
||
def test_enqueue_unknown_tid(self):
|
||
res = self.store.enqueue([9999999])
|
||
self.assertEqual(res["skipped_unknown"], 1)
|
||
|
||
def test_stats_and_states(self):
|
||
self.store.enqueue([7009126, 7303235])
|
||
self.assertEqual(self.store.stats()["pending"], 2)
|
||
self.store.claim_one(7009126)
|
||
self.store.finish(7009126, False, error="取址失败")
|
||
self.assertEqual(self.store.stats()["pending"], 2) # 失败后回到 pending（还有重试）
|
||
# 显式 backoff=0：验证"达到重试上限后落到 failed"（退避行为由 test_tasks 覆盖）
|
||
for _ in range(5):
|
||
if self.store.claim_one(7009126, max_attempts=3, backoff_sec=0):
|
||
self.store.finish(7009126, False, error="x")
|
||
self.assertEqual(self.store.stats()["failed"], 1)
|
||
|
||
def test_media_scan_requeues_missing_and_adopts(self):
|
||
"""媒体目录扫描（唯一的一致性入口）：done 但文件缺失 → 回队；磁盘上有文件 → 认领。"""
|
||
media = os.path.join(self.tmp, "media")
|
||
os.makedirs(media, exist_ok=True)
|
||
big = os.urandom(1024)
|
||
# 歌曲表里登记 done 但文件不存在 → 回退为待下载
|
||
self.store.keep_add([7009126])
|
||
self.store.keep_mark_done(7009126, len(big), os.path.join(media, "7009126.ts"))
|
||
res = self.store.keep_adopt_media(media, min_bytes=1, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 1)
|
||
self.assertEqual(self.store.q1("SELECT state FROM keeps WHERE tid=7009126")["state"], "want")
|
||
# 磁盘上有文件但歌曲表里还没下完 → 认领为 done（不再重复下载）
|
||
with open(os.path.join(media, "7303235.ts"), "wb") as fh:
|
||
fh.write(big)
|
||
self.store.keep_add([7303235])
|
||
res = self.store.keep_adopt_media(media, min_bytes=1, timeout_sec=5)
|
||
self.assertEqual(res["updated"], 1)
|
||
self.assertEqual(self.store.q1("SELECT state FROM keeps WHERE tid=7303235")["state"], "done")
|
||
|
||
|
||
class TestConfig(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.mkdtemp()
|
||
self.cfg = Config(os.path.join(self.tmp, "c.json"))
|
||
|
||
def test_concurrency_capped_at_10(self):
|
||
warns = self.cfg.update({"concurrency": 99})
|
||
self.assertEqual(self.cfg.get("concurrency"), 10)
|
||
self.assertTrue(any("10" in w for w in warns))
|
||
|
||
def test_warns_on_bare_ip_media_base(self):
|
||
warns = self.cfg.update({"public_media_base": "http://1.2.3.4/media"})
|
||
self.assertTrue(any("裸 IP" in w for w in warns))
|
||
|
||
def test_hosts_string_is_normalised(self):
|
||
self.cfg.update({"upstream_hosts": "http://a.example.com:6677, http://b.example.com"})
|
||
self.assertEqual(self.cfg.get("upstream_hosts"),
|
||
["http://a.example.com:6677", "http://b.example.com"])
|
||
|
||
def test_media_url(self):
|
||
self.cfg.update({"public_media_base": "http://ktv.example.com:6677/media"})
|
||
self.assertEqual(self.cfg.media_public_url(7009126), "http://ktv.example.com:6677/media/7009126.ts")
|
||
|
||
|
||
class TestLangNorm(unittest.TestCase):
|
||
def test_norm(self):
|
||
self.assertEqual(musesync.norm_lang("闽南"), "闽南语")
|
||
self.assertEqual(musesync.norm_lang("国语"), "国语")
|
||
self.assertEqual(musesync.norm_lang("夏天"), "其它")
|
||
self.assertEqual(musesync.norm_lang(""), "未标注")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|