Files
zijunchiang 87aa979abd 合并自建歌曲源支线(selfhost-source)+ 版本号按约定修正为 1.6.0
版本号:
- 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 例)
2026-10-14 02:06:28 +08:00

380 lines
19 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""设备身份(**伪装成一台安卓盒子**)自测:全程离线,不启动任何网络服务、不发 whoami。
覆盖需求里的五条:
1. 首次调用生成文件、第二次调用完全一致(持久化生效);
2. `regenerate()` 后 mac/sn 变化、模板仍在预设集合内、`_cid == "A_" + mac`、`androidid == mac`;
3. mac 合法性:12 位小写 hex、**单播**(第一个八位组最低位为 0)、OUI 属盒子厂商;
4. **一致性**:同一份身份构造 `i.php` / `music/do.php` / `whoami` 三个请求,mac/device/sn 同源;
5. `upstream_mac` 显式配置优先于自动身份(旧版本的抓包常量则不算显式配置)。
python3 -m unittest tests.test_identity -v
"""
import json
import os
import stat
import sys
import tempfile
import unittest
import urllib.parse
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ktv import admin, identity, resolver, whoami
from ktv.config import Config
from ktv.store import Store
from ktv.web import Request
def _req(method: str, path: str, body: bytes = b"") -> Request:
return Request(method, path, {}, {}, body)
class TestTemplates(unittest.TestCase):
"""模板本身必须像一台真盒子(自洽性由 `identity.validate()` 机器判定)。"""
def test_every_template_generates_a_valid_identity(self):
self.assertTrue(3 <= len(identity.TEMPLATES) <= 5, "内置模板 3~5 套")
for tpl in identity.TEMPLATES:
ident = identity.generate(tpl["preset"])
self.assertEqual([], identity.validate(ident), tpl["preset"])
self.assertEqual(tpl["preset"], ident["preset"])
self.assertEqual(tpl["_manuf"], ident["_manuf"])
self.assertEqual(tpl["_m"], ident["_m"])
# `_api` 与 `d_release` 必须自洽(9→28、11→30、12→31)
self.assertIn(ident["d_release"], ("9", "11", "12"))
self.assertEqual(identity.android_api(ident["d_release"]), ident["_api"])
# 真机指纹格式:品牌/产品/设备:版本/构建ID/增量:user/release-keys
fp = ident["d_fingerprint"]
self.assertTrue(fp.startswith(ident["_manuf"] + "/"), fp)
self.assertTrue(fp.endswith(":user/release-keys"), fp)
head, tail = fp.split(":", 1) # tail = 版本/构建ID/增量:user/release-keys
self.assertEqual(3, len(head.split("/")), fp) # 品牌/产品/设备
self.assertEqual(3, len(tail.split(":")[0].split("/")), fp) # 版本/构建ID/增量
# 抓包口径:androidid / _cid 与 mac 同源
self.assertEqual(ident["mac"], ident["androidid"])
self.assertEqual("A_" + ident["mac"], ident["_cid"])
# App/服务端共用口径的固定字段
self.assertEqual(4, ident["_vc"])
self.assertEqual(resolver.VN, ident["_vn"])
self.assertEqual("com.origjoy.local.ktv", ident["_pkn"])
def test_mac_is_unicast_hex_and_oui_is_a_real_box_vendor(self):
for _ in range(60):
ident = identity.generate()
mac = ident["mac"]
self.assertRegex(mac, r"^[0-9a-f]{12}$")
self.assertEqual(0, int(mac[:2], 16) & 0x01, f"组播地址:{mac}")
self.assertNotEqual("000000000000", mac)
self.assertIn(mac[:6], identity.OUIS)
self.assertRegex(ident["sn"], r"^[0-9a-f]{12}$")
def test_random_preset_and_unknown_preset(self):
for _ in range(20):
self.assertIn(identity.generate()["preset"], identity.PRESETS)
with self.assertLogs("ktv.identity", level="WARNING"):
ident = identity.generate("不存在的模板")
self.assertIn(ident["preset"], identity.PRESETS)
def test_validate_catches_bad_identities(self):
good = identity.generate("aml_s905x")
self.assertEqual([], identity.validate(good))
cases = {
"组播 mac": {**good, "mac": "011ec0abcdef", "androidid": "011ec0abcdef",
"_cid": "A_011ec0abcdef"},
"长度不对": {**good, "mac": "abc", "androidid": "abc", "_cid": "A_abc"},
"OUI 不认识": {**good, "mac": "aabbccddeeff", "androidid": "aabbccddeeff",
"_cid": "A_aabbccddeeff"},
"androidid 不同源": {**good, "androidid": "000000000000"},
"_cid 前缀不对": {**good, "_cid": good["mac"]},
"_api 与版本不自洽": {**good, "_api": 31},
"指纹结尾不像真机": {**good, "d_fingerprint": "Amlogic/a95x/a95x:9/x/y:eng/test-keys"},
"空机型": {**good, "_m": ""},
}
for name, broken in cases.items():
self.assertNotEqual([], identity.validate(broken), name)
self.assertNotEqual([], identity.validate(None))
class TestPersistence(unittest.TestCase):
"""生成一次、持久化、之后稳定不变。"""
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.path = os.path.join(self.tmp, identity.IDENTITY_FILENAME)
def _leftovers(self):
return [f for f in os.listdir(self.tmp) if ".tmp" in f]
def test_first_call_creates_file_second_call_is_identical(self):
first = identity.load_or_create(self.tmp)
self.assertTrue(os.path.isfile(self.path))
self.assertEqual([], identity.validate(first))
self.assertEqual(0o600, stat.S_IMODE(os.stat(self.path).st_mode), "身份文件按凭据对待(0600)")
second = identity.load_or_create(self.tmp)
self.assertEqual(first, second, "第二次必须复用文件,不能重新生成")
self.assertEqual([], self._leftovers(), "原子写不应留下临时文件")
def test_corrupt_file_is_regenerated(self):
with open(self.path, "w", encoding="utf-8") as fh:
fh.write("{ 这不是 JSON")
ident = identity.load_or_create(self.tmp)
self.assertEqual([], identity.validate(ident))
self.assertEqual(ident, identity.load(self.path))
def test_inconsistent_file_is_regenerated(self):
bad = identity.generate()
bad["_cid"] = "B_" + bad["mac"] # 破坏"三处同源"
identity.save(bad, self.path)
ident = identity.load_or_create(self.tmp)
self.assertEqual([], identity.validate(ident))
self.assertNotEqual(bad["mac"], ident["mac"])
def test_regenerate_changes_mac_and_sn(self):
before = identity.load_or_create(self.tmp)
after = identity.regenerate(self.tmp)
self.assertNotEqual(before["mac"], after["mac"])
self.assertNotEqual(before["sn"], after["sn"])
self.assertIn(after["preset"], identity.PRESETS, "模板必须还在预设集合内")
self.assertEqual("A_" + after["mac"], after["_cid"])
self.assertEqual(after["mac"], after["androidid"])
self.assertEqual([], identity.validate(after))
self.assertEqual(after, identity.load(self.path), "重新生成后必须已落盘")
self.assertEqual([], self._leftovers())
def test_regenerate_with_preset(self):
ident = identity.regenerate(self.tmp, preset="rk3328")
self.assertEqual("rk3328", ident["preset"])
self.assertEqual("Rockchip", ident["_manuf"])
self.assertEqual([], identity.validate(ident))
def test_custom_file_path(self):
custom = os.path.join(self.tmp, "sub", "my-identity.json")
ident = identity.load_or_create(self.tmp, path=custom)
self.assertTrue(os.path.isfile(custom))
self.assertEqual(ident, identity.load(custom))
self.assertFalse(os.path.isfile(self.path), "不该再去默认路径写一份")
class TestThreeRequestsShareOneIdentity(unittest.TestCase):
"""需求④:同一份身份 → `i.php` / `do.php` / `whoami` 三处的 mac/device/sn 完全同源。"""
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.cfg = Config(os.path.join(self.tmp, "c.json"))
self.cfg.update({"data_dir": self.tmp, "media_dir": os.path.join(self.tmp, "media")})
self.mgr = identity.DeviceIdentity(self.cfg)
self.ident = self.mgr.ensure()
self.assertIsNotNone(self.ident)
def _three_requests(self):
"""返回 (i.php 参数, do.php 参数, whoami 参数) —— 与真实请求同一套拼装函数。"""
mac, sn = resolver.Resolver(self.cfg).mac_sn()
token = "TOKEN"
i = dict(urllib.parse.parse_qsl(resolver.token_params(mac, sn, ts=1700000000)))
d = dict(urllib.parse.parse_qsl(
resolver.song_params(7009126, token, mac, sn, ts=1700000000)))
w = dict(whoami.build_query(self.cfg, boot=True, now_ms=1700000000000))
return i, d, w
def test_mac_device_sn_are_same_source(self):
i, d, w = self._three_requests()
mac, sn = self.ident["mac"], self.ident["sn"]
# i.php:`mac=<mac>_<sn>` 且 `sn=<sn>`
self.assertEqual(f"{mac}_{sn}", i["mac"])
self.assertEqual(sn, i["sn"])
# do.php:`device=<mac>_<sn>` 且 `sn=<sn>`
self.assertEqual(f"{mac}_{sn}", d["device"])
self.assertEqual(sn, d["sn"])
# whoami:`_mac`/`androidid`/`_cid` 三者同源,且与上面同一个 mac
self.assertEqual(mac, w["_mac"])
self.assertEqual(mac, w["androidid"])
self.assertEqual("A_" + mac, w["_cid"])
self.assertEqual(w["_mac"], w["androidid"])
def test_whoami_reports_the_fake_box_not_the_linux_host(self):
_i, _d, w = self._three_requests()
for key in ("_m", "_manuf", "d_name", "d_device", "d_board", "d_hardware", "d_incremental"):
self.assertEqual(self.ident[key], w[key], key)
self.assertEqual(self.ident["d_fingerprint"], w["d_fingerprint"])
self.assertEqual(self.ident["d_release"], w["d_release"])
self.assertEqual(str(self.ident["_api"]), w["_api"])
self.assertEqual(self.ident["ktv_id"], w["ktv_id"])
self.assertTrue(w["d_fingerprint"].endswith(":user/release-keys"))
# UA 也按同一台盒子生成(型号/版本来自身份)
ua = whoami.user_agent(self.cfg)
self.assertIn(self.ident["_m"], ua)
self.assertIn(f"Android {self.ident['d_release']}", ua)
# 24 个字段一个不少(顺序由 whoami.PARAM_ORDER 保证)
self.assertEqual(24, len(whoami.build_query(self.cfg, boot=False)))
# `sn` 不在 whoami 的 24 个字段里(抓包如此),它由 i.php / do.php 使用
self.assertNotIn("sn", w)
def test_regenerate_moves_all_three_together(self):
_i0, _d0, w0 = self._three_requests()
self.mgr.regenerate("aml_s905x3")
ident = self.mgr.identity
i, d, w = self._three_requests()
mac, sn = ident["mac"], ident["sn"]
self.assertNotEqual(w0["_mac"], w["_mac"], "重新生成后 mac 必须变化")
self.assertEqual("aml_s905x3", ident["preset"])
self.assertEqual(mac, i["mac"].split("_")[0])
self.assertEqual(sn, i["mac"].split("_")[1])
self.assertEqual(mac, d["device"].split("_")[0])
self.assertEqual(sn, d["device"].split("_")[1])
self.assertEqual(mac, w["_mac"])
self.assertEqual("A_" + mac, w["_cid"])
def test_explicit_whoami_api_keeps_release_consistent(self):
"""显式改 `_api` 时,`d_release` 与指纹的版本段必须跟着改(不能自相矛盾)。"""
self.cfg.update({"whoami_api": 31}) # 这台盒子模板是 Android 9(api 28)
w = dict(whoami.build_query(self.cfg))
self.assertEqual("31", w["_api"])
self.assertEqual("12", w["d_release"])
self.assertIn(":12/", w["d_fingerprint"])
self.assertTrue(w["d_fingerprint"].endswith(":user/release-keys"))
def test_auto_disabled_reports_real_host(self):
"""关掉自动身份(且无身份文件)时走兜底:上报真实本机机型,但绝不生成文件。"""
tmp = tempfile.mkdtemp()
cfg = Config(os.path.join(tmp, "c.json"))
cfg.update({"data_dir": tmp, "device_identity_auto": False,
"media_dir": os.path.join(tmp, "media")})
mgr = identity.DeviceIdentity(cfg)
self.assertIsNone(mgr.ensure())
self.assertEqual("disabled", mgr.origin)
self.assertFalse(os.path.isfile(os.path.join(tmp, identity.IDENTITY_FILENAME)))
w = dict(whoami.build_query(cfg))
self.assertEqual("", w["_mac"]) # 没有身份就退回配置值(默认空)
self.assertTrue(w["d_fingerprint"].endswith(":user/release-keys"))
class TestExplicitConfigWins(unittest.TestCase):
"""需求⑤:`upstream_mac` 显式配置时优先于自动身份;旧版本的抓包常量不算显式。"""
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.cfg_path = os.path.join(self.tmp, "c.json")
def test_explicit_upstream_mac_wins(self):
cfg = Config(self.cfg_path)
cfg.update({"data_dir": self.tmp, "upstream_mac": "aabbccddeeff",
"upstream_sn": "001122334455"})
mgr = identity.DeviceIdentity(cfg)
ident = mgr.ensure()
self.assertIsNotNone(ident, "身份仍然生成,只是不覆盖显式配置")
self.assertEqual("aabbccddeeff", cfg.get("upstream_mac"))
self.assertEqual("001122334455", cfg.get("upstream_sn"))
st = mgr.status()
self.assertEqual("explicit", st["effective"]["mac_source"])
self.assertEqual("explicit", st["effective"]["sn_source"])
self.assertEqual("explicit", mgr.origin)
# whoami 也跟着显式配置走(三处仍然同源,只是源头是用户配置)
mac, sn = resolver.Resolver(cfg).mac_sn()
self.assertEqual(("aabbccddeeff", "001122334455"), (mac, sn))
w = dict(whoami.build_query(cfg))
self.assertEqual("aabbccddeeff", w["_mac"])
self.assertEqual("aabbccddeeff", w["androidid"])
self.assertEqual("A_aabbccddeeff", w["_cid"])
self.assertEqual(ident["_m"], w["_m"], "机型仍来自盒子身份")
def test_legacy_capture_constants_are_not_explicit(self):
"""旧部署的 server-config.json 里写着抓包常量 → 应被自动身份覆盖(升级即生效)。"""
with open(self.cfg_path, "w", encoding="utf-8") as fh:
json.dump({"data_dir": self.tmp, "upstream_mac": "800b510bf4fd",
"upstream_sn": "2d7bb4a7d7f7"}, fh)
cfg = Config(self.cfg_path)
self.assertFalse(cfg.is_explicit("upstream_mac"))
self.assertFalse(cfg.is_explicit("upstream_sn"))
ident = identity.DeviceIdentity(cfg).ensure()
self.assertEqual(ident["mac"], cfg.get("upstream_mac"))
self.assertEqual(ident["sn"], cfg.get("upstream_sn"))
self.assertNotEqual("800b510bf4fd", cfg.get("upstream_mac"))
def test_auto_values_are_not_persisted(self):
"""自动身份填进配置的值**不落盘**;否则下次它就成了"显式配置",重生成也换不掉。"""
cfg = Config(self.cfg_path)
cfg.update({"data_dir": self.tmp})
ident = identity.DeviceIdentity(cfg).ensure()
cfg.save()
with open(self.cfg_path, "r", encoding="utf-8") as fh:
saved = json.load(fh)
self.assertNotIn("upstream_mac", saved)
self.assertNotIn("upstream_sn", saved)
# 重启一次:仍然跟随身份文件(mac 不变)
cfg2 = Config(self.cfg_path)
self.assertFalse(cfg2.is_explicit("upstream_mac"))
ident2 = identity.DeviceIdentity(cfg2).ensure()
self.assertEqual(ident["mac"], ident2["mac"])
self.assertEqual(ident["mac"], cfg2.get("upstream_mac"))
# 用户显式清空 → 交还给自动身份
cfg2.update({"upstream_mac": ""})
self.assertFalse(cfg2.is_explicit("upstream_mac"))
cfg2.update({"upstream_mac": "aabbccddeeff"})
self.assertTrue(cfg2.is_explicit("upstream_mac"))
class TestDeviceApi(unittest.TestCase):
"""管理接口:`GET /api/device` 与 `POST /api/device/regenerate`。"""
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.cfg = Config(os.path.join(self.tmp, "c.json"))
self.cfg.update({"data_dir": self.tmp, "media_dir": os.path.join(self.tmp, "media")})
self.store = Store(os.path.join(self.tmp, "t.db"))
self.addCleanup(self.store.conn.close)
self.mgr = identity.DeviceIdentity(self.cfg, self.store)
self.mgr.ensure()
self.router = admin.build_router(self.cfg, self.store, None, device_identity=self.mgr)
def test_get_device_status(self):
st = json.loads(self.router.dispatch(_req("GET", "/api/device")).body)
self.assertTrue(st["ok"])
self.assertEqual(self.mgr.path(), st["file"])
self.assertTrue(st["exists"])
self.assertEqual(self.mgr.identity["mac"], st["identity"]["mac"])
self.assertEqual(self.mgr.identity["preset"], st["preset"])
self.assertEqual([], st["problems"])
self.assertGreaterEqual(len(st["presets"]), 3)
self.assertEqual(self.mgr.identity["mac"], st["effective"]["upstream_mac"])
self.assertEqual("identity", st["effective"]["mac_source"])
def test_regenerate_refreshes_config_and_logs_event(self):
old = self.mgr.identity["mac"]
body = json.dumps({"preset": "aml_s905x3"}).encode()
res = json.loads(self.router.dispatch(
_req("POST", "/api/device/regenerate", body)).body)
self.assertTrue(res["ok"])
self.assertEqual("aml_s905x3", res["identity"]["preset"])
self.assertNotEqual(old, res["identity"]["mac"])
# ① 落盘;② 内存配置立刻刷新(i.php / do.php 下一轮就用新的)
self.assertEqual(res["identity"], identity.load(self.mgr.path()))
self.assertEqual(res["identity"]["mac"], self.cfg.get("upstream_mac"))
self.assertEqual(res["identity"]["sn"], self.cfg.get("upstream_sn"))
# ③ whoami 下一轮用的也是新身份(whoami 每轮现读配置)
w = dict(whoami.build_query(self.cfg))
self.assertEqual(res["identity"]["mac"], w["_mac"])
self.assertEqual(res["identity"]["_m"], w["_m"])
# ④ 事件日志
texts = " ".join(e["text"] for e in self.store.events(20))
self.assertIn("设备身份重新生成", texts)
def test_endpoint_is_safe_without_manager(self):
router = admin.build_router(self.cfg, self.store, None)
st = router.dispatch(_req("GET", "/api/device"))
self.assertEqual(503, st.status)
regen = router.dispatch(_req("POST", "/api/device/regenerate", b"{}"))
self.assertEqual(503, regen.status)
def test_unknown_preset_does_not_break_the_api(self):
# 模板名写错只该记一条警告并回退随机(这里顺手把它从测试输出里收走)
with self.assertLogs("ktv.identity", level="WARNING"):
res = json.loads(self.router.dispatch(_req(
"POST", "/api/device/regenerate", json.dumps({"preset": "不存在"}).encode())).body)
self.assertTrue(res["ok"])
self.assertIn(res["identity"]["preset"], identity.PRESETS)
if __name__ == "__main__":
unittest.main(verbosity=2)