#!/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)