#!/usr/bin/env python3 """App 令牌（`GET /token`）与**双站点反代**（API 站 / 媒体站）自测。 覆盖： ① `/token`：随机且每次不同、`expires_in=600`、`Cache-Control: no-store`、 TTL 过期后校验失败（注入时钟，不真等 10 分钟）、绑定客户端 IP、单次令牌、 可选持久化（重启仍有效）、每分钟签发限流、聚合日志（按分钟一条）； ② 校验与兼容：`app_token_enforce=false` 时老 App 不带令牌照常可用、带**错误**令牌报 401； `enforce=true` 时没令牌 → 401、带合法令牌 → 通过；`/i.php` 同口径； 老协议自己的 `token=local` **不算** App 令牌（否则已发布 App 会被全线 401）； ③ `/push_download`：合法 App 令牌 **或** 合法 `push_token` 任一即可，都没有 → 401； ④ 宿主隔离：媒体宿主只能下文件、API 宿主不能批量下载、**不可信代理头一律忽略**（防伪造）； ⑤ HTTP 层端到端（真起 127.0.0.1 端口）：取令牌 → 带 `X-App-Token` 调 do.php 成功； `enforce=true` 不带令牌 401；伪造 `X-Forwarded-Host` 走媒体站 → 200 + Range 206、 走 do.php → 404；不可信代理头伪造宿主无效； ⑥ `deploy/iis/` 两份站点配置静态校验（可解析、各管各的、都不含管理端口）。 不联网、不起外网连接（socket 用例只绑 127.0.0.1 的临时端口）。 """ import http.client import json import os import re import shutil import sys import tempfile import unittest import urllib.parse import xml.etree.ElementTree as ET sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ktv import apptoken, proxy, service, web from ktv.config import Config from ktv.store import Store from ktv.web import Request HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) IIS_DIR = os.path.join(os.path.dirname(HERE), "deploy", "iis") API_CFG = os.path.join(IIS_DIR, "api-site", "web.config") MEDIA_CFG = os.path.join(IIS_DIR, "media-site", "web.config") TID = 7300123 PUSH_URL = f"http://cdn.example.com/480p/{TID}.ts?token=abc&t=1700000000" def req(method, path, query=None, body=b"", headers=None, client_ip="", peer_ip="", via_proxy=False, proto="", host="", xff=""): """构造 Request；反代场景用 `via_proxy=True` + `proto/host/xff`（= web.py 注入的口径）。""" return Request(method, path, query or {}, headers or {}, body, client_ip=client_ip, peer_ip=peer_ip or client_ip, via_proxy=via_proxy, proxy_source="x-forwarded-for" if (via_proxy and xff) else "tcp", forwarded_proto=proto, forwarded_host=host, forwarded_for=xff) def form(**kv) -> bytes: return urllib.parse.urlencode(kv).encode() def via_proxy_req(cfg, method, path, query=None, body=b"", peer="127.0.0.1", xff="203.0.113.9", xfh="", headers=None, client_ip=None): """按**真实认定链路**（proxy.resolve）构造一条"经可信代理到达"的请求。""" raw = {"X-Forwarded-For": xff} if xfh: raw["X-Forwarded-Host"] = xfh info = proxy.resolve(peer, raw, cfg) hdrs = {"Host": "127.0.0.1:6677", "X-Forwarded-For": xff} if xfh: hdrs["X-Forwarded-Host"] = xfh hdrs.update(headers or {}) return req(method, path, query, body, hdrs, client_ip=info.ip if client_ip is None else client_ip, peer_ip=info.peer_ip, via_proxy=info.via_proxy, proto=info.proto, host=info.host, xff=info.xff) class TokenCase(unittest.TestCase): """公共脚手架：一份配置 + 一个库 + 服务端口路由（含守卫）。""" def setUp(self): self.tmp = tempfile.mkdtemp() self.media = os.path.join(self.tmp, "media") os.makedirs(self.media, exist_ok=True) self.cfg = Config(os.path.join(self.tmp, "c.json")) self.cfg.update({"data_dir": self.tmp, "media_dir": self.media, "public_media_base": "https://md.kasugano.cn:4433/media"}) self.store = Store(os.path.join(self.tmp, "t.db")) self.router = service.build_router(self.cfg, self.store, None) self.tokens = apptoken.registry_for(self.cfg, self.store) # 注入时钟：TTL / 限流窗口 / 分钟聚合都按它走，测试不真等 10 分钟 self.now = [1_700_000_000.0] self.tokens._clock = lambda: self.now[0] def tearDown(self): try: self.store.conn.close() except Exception: pass shutil.rmtree(self.tmp, ignore_errors=True) # ── 小工具 ── def call(self, r): return self.router.dispatch(r) def get_token(self, ip="", method="GET"): resp = self.call(req(method, "/token", client_ip=ip)) return resp, json.loads(resp.body) def ready_song(self, tid=TID, size=7 << 20): """把某首歌做成"已下载完成"（do.php 才可能回 code:200）。""" self.store.import_library([(tid, "测试歌", "国语", "", "歌手", f"{tid}.ts", 10, "")], "v") self.store.enqueue([tid]) self.store.claim_one(tid) path = os.path.join(self.media, f"{tid}.ts") with open(path, "wb") as fh: fh.write(b"\x47" * 300) self.store.finish(tid, True, bytes_=size, path=path) return tid def do_php(self, tid=TID, **kw): return self.call(req("GET", "/music/do.php", {"musicno": str(tid)}, **kw)) def advance(self, seconds): self.now[0] += float(seconds) # ══════════════════════ ① /token 本身 ══════════════════════ class TestTokenEndpoint(TokenCase): def test_returns_random_token_with_ttl_and_no_store(self): """随机、每次不同、expires_in=600、禁止缓存。""" self.assertEqual(self.cfg.get("app_token_ttl_sec"), 600) resp, d = self.get_token(ip="203.0.113.9") self.assertEqual(resp.status, 200) self.assertTrue(d["ok"]) self.assertEqual(d["expires_in"], 600) self.assertEqual(d["expires_at"], int(self.now[0]) + 600) self.assertEqual(resp.headers.get("Cache-Control"), "no-store") # 32~48 位、URL-safe（hex/urlsafe 都在这个字符集里） self.assertRegex(d["token"], r"^[A-Za-z0-9_\-=]{32,48}$") self.assertGreaterEqual(len(d["token"]), 32) self.assertLessEqual(len(d["token"]), 48) seen = {d["token"]} for _ in range(20): _r, d2 = self.get_token(ip="203.0.113.9") self.assertNotIn(d2["token"], seen, "令牌必须每次不同") seen.add(d2["token"]) # POST 也接受（有些客户端不方便发 GET） rp, dp = self.get_token(ip="203.0.113.9", method="POST") self.assertEqual(rp.status, 200) self.assertTrue(dp["ok"]) def test_ttl_expiry_uses_injected_clock(self): """TTL 过期后校验失败 —— 用注入时钟，不真等。""" self.cfg.update({"app_token_ttl_sec": 30}) # 最小合法值 _r, d = self.get_token(ip="203.0.113.9") tok = d["token"] self.assertEqual(d["expires_in"], 30) self.ready_song() self.assertEqual(self.do_php(headers={"X-App-Token": tok}, client_ip="203.0.113.9").status, 200) self.advance(31) resp = self.do_php(headers={"X-App-Token": tok}, client_ip="203.0.113.9") self.assertEqual(resp.status, 401) self.assertEqual(json.loads(resp.body), {"ok": False, "error": apptoken.AUTH_ERROR}) def test_ttl_clamped_config(self): warns = self.cfg.update({"app_token_ttl_sec": 3}) self.assertEqual(self.cfg.get("app_token_ttl_sec"), 30) self.assertTrue(any("有效期" in w for w in warns)) self.cfg.update({"app_token_ttl_sec": 10 ** 9}) self.assertEqual(self.cfg.get("app_token_ttl_sec"), 86400) def test_bind_ip_rejects_other_client(self): """令牌绑定签发时的客户端 IP；换 IP 用同一令牌 → 401。""" self.assertTrue(self.cfg.get("app_token_bind_ip")) _r, d = self.get_token(ip="203.0.113.9") tok = d["token"] self.ready_song() self.assertEqual(self.tokens.verify(tok, "203.0.113.9"), True) self.assertEqual(self.tokens.verify(tok, "203.0.113.10"), False) self.assertEqual(self.do_php(headers={"X-App-Token": tok}, client_ip="203.0.113.10").status, 401) # 关掉绑定 → 换 IP 也放行 self.cfg.update({"app_token_bind_ip": False}) self.assertEqual(self.tokens.verify(tok, "203.0.113.10"), True) def test_single_use_consumes_token(self): """`app_token_single_use=true` 用过即失效（默认必须是 False：取址是多次调用）。""" _r, d = self.get_token(ip="203.0.113.9") tok = d["token"] self.assertTrue(self.tokens.verify(tok, "203.0.113.9")) self.assertTrue(self.tokens.verify(tok, "203.0.113.9"), "默认不是单次，应可多次使用") self.cfg.update({"app_token_single_use": True}) _r, d2 = self.get_token(ip="203.0.113.9") tok2 = d2["token"] self.assertTrue(self.tokens.verify(tok2, "203.0.113.9")) self.assertFalse(self.tokens.verify(tok2, "203.0.113.9")) def test_optional_persistence(self): """`app_token_persist=true` → 落 meta 表，重启（新实例）后仍有效；默认纯内存不落库。""" self.assertFalse(self.cfg.get("app_token_persist")) _r, d = self.get_token(ip="203.0.113.9") fresh = apptoken.TokenService(self.cfg, self.store, clock=lambda: self.now[0]) self.assertFalse(fresh.verify(d["token"], "203.0.113.9"), "默认纯内存：重启即失效") self.cfg.update({"app_token_persist": True}) _r, d2 = self.get_token(ip="203.0.113.9") self.assertTrue(self.store.get_meta(apptoken.META_KEY, ""), "应该落了一行 meta") fresh2 = apptoken.TokenService(self.cfg, self.store, clock=lambda: self.now[0]) self.assertTrue(fresh2.verify(d2["token"], "203.0.113.9"), "持久化后重启应仍有效") # 过期后不该被"复活" self.advance(601) fresh3 = apptoken.TokenService(self.cfg, self.store, clock=lambda: self.now[0]) self.assertFalse(fresh3.verify(d2["token"], "203.0.113.9")) def test_rate_limit_per_minute_per_ip(self): """同一来源 IP 每分钟签发数受限（默认 60，防止被刷）。""" self.cfg.update({"token_rate_limit_per_min": 2}) self.assertEqual(self.get_token(ip="203.0.113.9")[0].status, 200) self.assertEqual(self.get_token(ip="203.0.113.9")[0].status, 200) third = self.get_token(ip="203.0.113.9")[0] self.assertEqual(third.status, 429) self.assertEqual(third.headers.get("Cache-Control"), "no-store") self.assertIn("获取令牌过于频繁", json.loads(third.body)["error"]) # 另一个 IP 不受影响 self.assertEqual(self.get_token(ip="203.0.113.10")[0].status, 200) # 窗口滑动 60 秒后恢复 self.advance(61) self.assertEqual(self.get_token(ip="203.0.113.9")[0].status, 200) # 0 = 关闭限流 self.cfg.update({"token_rate_limit_per_min": 0}) for _ in range(5): self.assertEqual(self.get_token(ip="203.0.113.9")[0].status, 200) def test_events_are_aggregated_per_minute(self): """签发/校验失败按分钟聚合成**一条**（这个项目很在意日志量）。""" for _ in range(3): self.get_token(ip="203.0.113.9") for _ in range(2): self.tokens.verify("bogus-token", "203.0.113.9") self.advance(120) # 跨过分钟边界 self.get_token(ip="203.0.113.10") # 触发汇总 texts = [e["text"] for e in self.store.events(50)] issued = [t for t in texts if "App 令牌：签发 3 个" in t] failed = [t for t in texts if "App 令牌：校验失败 2 次" in t] self.assertEqual(len(issued), 1, texts) self.assertEqual(len(failed), 1, texts) self.assertIn("203.0.113.9", failed[0]) # 成功校验**不逐笔**记录：连打 10 次也没有"成功"日志 tok = self.get_token(ip="203.0.113.9")[1]["token"] for _ in range(10): self.tokens.verify(tok, "203.0.113.9") self.assertFalse(any("令牌校验成功" in t for t in [e["text"] for e in self.store.events(50)])) def test_status_endpoint(self): self.get_token(ip="203.0.113.9") self.tokens.verify("nope", "203.0.113.9") d = json.loads(self.call(req("GET", "/token/status")).body) self.assertTrue(d["ok"]) self.assertEqual(d["issued"], 1) self.assertEqual(d["invalid"], 1) self.assertEqual(d["ttl_sec"], 600) self.assertEqual(d["paths"], ["/music/do.php", "/push_download"]) self.assertIn("pending_log", d) # ══════════════════════ ② 校验与兼容（红线） ══════════════════════ class TestTokenEnforce(TokenCase): def do_php_with_token_param(self, value): """老 App 的口径：令牌放在 query 的 `token` 里（= i.php 发给它的那个）。""" return self.call(req("GET", "/music/do.php", {"musicno": str(TID), "token": value})) def test_disabled_by_default_old_app_unaffected(self): """**红线**：默认不强制 —— 老 App（不带令牌）必须照常可用。""" self.assertFalse(self.cfg.get("app_token_enforce")) self.ready_song() self.assertEqual(self.do_php().status, 200) # 不带令牌 self.assertEqual(json.loads(self.do_php().body)["code"], 200) # 老协议自己的 token 参数（i.php 发的那个）不是 App 令牌 —— 不能被当成"错误令牌" self.assertEqual(self.do_php_with_token_param("local").status, 200) i = self.call(req("GET", "/i.php", {"mac": "x"})) self.assertEqual(i.status, 200) self.assertEqual(json.loads(i.body)["code"], 200) def test_wrong_token_rejected_even_when_not_enforced(self): """带了**错误**令牌必须报 401（避免"以为生效了"）。""" self.ready_song() resp = self.do_php_with_token_param("definitely-wrong") self.assertEqual(resp.status, 401) self.assertEqual(json.loads(resp.body), {"ok": False, "error": apptoken.AUTH_ERROR}) resp2 = self.call(req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"X-App-Token": "definitely-wrong"})) self.assertEqual(resp2.status, 401) # 过期令牌同样 401 self.cfg.update({"app_token_ttl_sec": 30}) tok = self.get_token(ip="")[1]["token"] self.advance(31) self.assertEqual(self.call(req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"X-App-Token": tok})).status, 401) def test_valid_token_passes_when_not_enforced(self): self.ready_song() tok = self.get_token(ip="203.0.113.9")[1]["token"] r = self.call(req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"X-App-Token": tok}, client_ip="203.0.113.9")) self.assertEqual(r.status, 200) self.assertEqual(json.loads(r.body)["code"], 200) # query 里的 App 令牌（非老协议令牌）也认 r2 = self.call(req("GET", "/music/do.php", {"musicno": str(TID), "token": tok}, client_ip="203.0.113.9")) self.assertEqual(r2.status, 200) def test_enforced_requires_token(self): self.ready_song() self.cfg.update({"app_token_enforce": True}) resp = self.do_php() self.assertEqual(resp.status, 401) self.assertEqual(json.loads(resp.body), {"ok": False, "error": apptoken.AUTH_ERROR}) tok = self.get_token(ip="203.0.113.9")[1]["token"] self.assertEqual(self.call(req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"X-App-Token": tok}, client_ip="203.0.113.9")).status, 200) # query 里的老协议令牌 ≠ App 令牌：强制模式下仍要 401（否则等于没有强制） self.assertEqual(self.do_php_with_token_param("local").status, 401) def test_i_php_checked_when_token_present(self): """`/i.php` 默认不在 app_token_paths 里（不强制），但"带了就校验"。""" self.assertEqual(self.call(req("GET", "/i.php", {"mac": "x"})).status, 200) self.assertEqual(self.call(req("GET", "/i.php", {"mac": "x"}, headers={"X-App-Token": "bad"})).status, 401) tok = self.get_token(ip="")[1]["token"] self.assertEqual(self.call(req("GET", "/i.php", {"mac": "x"}, headers={"X-App-Token": tok})).status, 200) def test_custom_paths(self): self.cfg.update({"app_token_paths": "/i.php", "app_token_enforce": True}) self.assertEqual(self.call(req("GET", "/i.php", {"mac": "x"})).status, 401) # do.php 不在受管路径里了 → 不强制 self.assertEqual(self.do_php().status, 200) self.assertIn("/i.php", self.tokens.paths()) class TestPushDownloadAuth(TokenCase): """`/push_download`：合法 App 令牌 **或** 合法 push_token 任一即可。""" def push(self, headers=None, body=None, client_ip="203.0.113.9"): payload = body if body is not None else form(tid=TID, url=PUSH_URL) return self.call(req("POST", "/push_download", {}, payload, headers or {}, client_ip=client_ip)) def test_app_token_accepted(self): self.cfg.update({"app_token_enforce": True}) tok = self.get_token(ip="203.0.113.9")[1]["token"] resp = self.push(headers={"X-App-Token": tok}) self.assertEqual(resp.status, 200, resp.body) self.assertTrue(json.loads(resp.body)["ok"]) def test_push_token_still_accepted(self): """原有 push_token 通道必须继续通过（老 App / 脚本不受影响）。""" resp = self.push(body=form(tid=TID, url=PUSH_URL, token="local")) self.assertEqual(resp.status, 200, resp.body) resp2 = self.push(headers={"X-Push-Token": "local"}) self.assertEqual(resp2.status, 200, resp2.body) def test_no_credential_is_401(self): resp = self.push() self.assertEqual(resp.status, 401) self.assertEqual(json.loads(resp.body), {"ok": False, "error": "鉴权失败"}) # 连 enforce=true 时也是 401（走 push 自己的那条判定） self.cfg.update({"app_token_enforce": True}) self.assertEqual(self.push().status, 401) def test_wrong_credentials_are_401(self): self.assertEqual(self.push(headers={"X-Push-Token": "wrong"}).status, 401) self.assertEqual(self.push(body=form(tid=TID, url=PUSH_URL, token="wrong")).status, 401) self.assertEqual(self.push(headers={"X-App-Token": "wrong"}).status, 401) # ══════════════════════ ④ 宿主隔离（双站点） ══════════════════════ class TestMediaHostIsolation(TokenCase): def setUp(self): super().setUp() self.ready_song() with open(os.path.join(self.media, f"{TID}.ts"), "wb") as fh: fh.write(b"\x47" * 300) # 300 字节，Range 断言用 self.cfg.update({"media_hosts": ["md.kasugano.cn"], "api_hosts": ["mdmuse.kasugano.cn"], "trust_proxy_headers": True, "trusted_proxies": ["127.0.0.1"]}) def hit(self, path, xfh, **kw): query = {"musicno": str(TID)} if path.startswith("/music/") else None return self.call(via_proxy_req(self.cfg, "GET", path, query, xfh=xfh, **kw)) def test_media_host_can_download(self): r = self.hit(f"/media/{TID}.ts", "md.kasugano.cn:4433") self.assertEqual(r.status, 200, r.body) self.assertEqual(r.headers.get("Accept-Ranges"), "bytes") r2 = self.hit(f"/media/{TID}.ts", "md.kasugano.cn:4433", headers={"Range": "bytes=100-199"}) self.assertEqual(r2.status, 206) self.assertEqual(r2.headers.get("Content-Range"), "bytes 100-199/300") self.assertEqual(r2.file_len, 100) def test_media_host_is_port_tolerant(self): """配置写不带端口，头里带 :4433 必须命中（反之亦然）。""" self.assertEqual(self.hit(f"/media/{TID}.ts", "md.kasugano.cn:4433").status, 200) self.assertEqual(self.hit(f"/media/{TID}.ts", "md.kasugano.cn").status, 200) self.cfg.update({"media_hosts": ["MD.Kasugano.CN:4433"]}) self.assertEqual(self.hit(f"/media/{TID}.ts", "md.kasugano.cn").status, 200) self.assertEqual(self.hit(f"/media/{TID}.ts", "md.kasugano.cn:4433").status, 200) def test_media_host_cannot_call_api(self): self.assertEqual(self.hit("/music/do.php", "md.kasugano.cn:4433").status, 404) self.assertEqual(self.hit("/healthz", "md.kasugano.cn:4433").status, 404) self.assertEqual(self.hit("/i.php", "md.kasugano.cn:4433").status, 404) self.assertEqual(self.hit("/token", "md.kasugano.cn:4433").status, 404) # 404 的响应体与"路径不存在"长得一样（不透露隔离规则） body = json.loads(self.hit("/healthz", "md.kasugano.cn:4433").body) self.assertEqual(body, {"error": "not found", "path": "/healthz"}) def test_api_host_cannot_download(self): self.assertEqual(self.hit(f"/media/{TID}.ts", "mdmuse.kasugano.cn").status, 404) self.assertEqual(self.hit("/music/do.php", "mdmuse.kasugano.cn").status, 200) self.assertEqual(self.hit("/healthz", "mdmuse.kasugano.cn").status, 200) def test_untrusted_forwarded_host_is_ignored(self): """**防伪造**：不可信对端塞 X-Forwarded-Host 一律不采信（只看 Host 头）。""" self.cfg.update({"trusted_proxies": ["192.168.100.10"]}) # 127.0.0.1 不再可信 forged = req("GET", f"/media/{TID}.ts", headers={"Host": "127.0.0.1:6677", "X-Forwarded-Host": "md.kasugano.cn:4433"}, client_ip="127.0.0.1") self.assertEqual(self.call(forged).status, 404, "伪造的媒体宿主不能下文件") forged2 = req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"Host": "127.0.0.1:6677", "X-Forwarded-Host": "md.kasugano.cn:4433"}, client_ip="127.0.0.1") self.assertEqual(self.call(forged2).status, 200, "不可信头不该把 API 也封掉") # 开关整体关掉时同样不采信 self.cfg.update({"trust_proxy_headers": False}) self.assertEqual(self.call(forged).status, 404) def test_host_falls_back_to_host_header_when_untrusted(self): """不可信时退回 Host 头：直接以媒体域名访问（没经代理）也能下文件。""" self.cfg.update({"trusted_proxies": ["192.168.100.10"]}) r = req("GET", f"/media/{TID}.ts", headers={"Host": "md.kasugano.cn:4433"}, client_ip="203.0.113.9") self.assertEqual(self.call(r).status, 200) r2 = req("GET", "/music/do.php", {"musicno": str(TID)}, headers={"Host": "md.kasugano.cn:4433"}, client_ip="203.0.113.9") self.assertEqual(self.call(r2).status, 404) def test_disabled_when_media_hosts_empty(self): self.cfg.update({"media_hosts": [], "api_hosts": ["mdmuse.kasugano.cn"]}) self.assertEqual(self.hit(f"/media/{TID}.ts", "mdmuse.kasugano.cn").status, 200) self.assertEqual(self.hit("/music/do.php", "anything.example.com").status, 200) def test_client_ip_reports_host_decision(self): r = self.hit("/api/client_ip", "md.kasugano.cn:4433") d = json.loads(r.body) self.assertEqual(d["host"], "md.kasugano.cn:4433") self.assertEqual(d["host_source"], "x-forwarded-host") self.assertTrue(d["media_host_match"]) self.assertFalse(d["api_host_match"]) self.assertEqual(d["media_hosts"], ["md.kasugano.cn"]) d2 = json.loads(self.hit("/api/client_ip", "mdmuse.kasugano.cn").body) self.assertFalse(d2["media_host_match"]) self.assertTrue(d2["api_host_match"]) def test_config_warnings(self): warns = self.cfg.update({"media_hosts": "md.kasugano.cn, 不是域名", "trust_proxy_headers": False}) self.assertEqual(self.cfg.get("media_hosts"), ["md.kasugano.cn", "不是域名"]) self.assertTrue(any("trust_proxy_headers" in w for w in warns)) self.assertTrue(any("看不懂" in w for w in warns)) # ══════════════════════ ⑤ HTTP 层端到端 ══════════════════════ class TestOverTheWire(unittest.TestCase): """真起 127.0.0.1 端口：令牌 + 双站点判定 + Range/206 全链路。""" def setUp(self): self.tmp = tempfile.mkdtemp() self.media = os.path.join(self.tmp, "media") os.makedirs(self.media, exist_ok=True) self.cfg = Config(os.path.join(self.tmp, "c.json")) self.cfg.update({"data_dir": self.tmp, "media_dir": self.media, "public_media_base": "https://md.kasugano.cn:4433/media", "trust_proxy_headers": True, "trusted_proxies": ["127.0.0.1"], "media_hosts": ["md.kasugano.cn"], "api_hosts": ["mdmuse.kasugano.cn"]}) self.store = Store(os.path.join(self.tmp, "t.db")) self.store.import_library([(TID, "测试歌", "国语", "", "歌手", f"{TID}.ts", 1, "")], "v") with open(os.path.join(self.media, f"{TID}.ts"), "wb") as fh: fh.write(b"\x47" * 300) self.router = service.build_router(self.cfg, self.store, None) self.srv = web.start_server("127.0.0.1", 0, self.router, cfg=self.cfg, backlog=32) self.port = self.srv.server_address[1] def tearDown(self): try: self.srv.shutdown() self.srv.server_close() except Exception: pass try: self.store.conn.close() except Exception: pass shutil.rmtree(self.tmp, ignore_errors=True) def get(self, path, headers=None, method="GET"): c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) try: c.request(method, path, headers=headers or {}) r = c.getresponse() return r.status, dict((k.lower(), v) for k, v in r.getheaders()), r.read() finally: c.close() def token_for(self, client_ip): st, hdrs, body = self.get("/token", {"X-Forwarded-For": client_ip}) self.assertEqual(st, 200, body) self.assertEqual(hdrs.get("cache-control"), "no-store") return json.loads(body)["token"] def test_full_app_token_flow(self): tok = self.token_for("203.0.113.9") # ① 带 X-App-Token 调 do.php：通过（歌曲没下载完 → JSON code:404，HTTP 仍是 200） st, _h, body = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.9", "X-App-Token": tok, "X-Forwarded-Host": "mdmuse.kasugano.cn"}) self.assertEqual(st, 200, body) self.assertIn(json.loads(body)["code"], (200, 404)) # ② 换了客户端 IP（同一令牌）→ 401 st2, _h2, body2 = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.10", "X-App-Token": tok}) self.assertEqual(st2, 401, body2) self.assertEqual(json.loads(body2)["error"], apptoken.AUTH_ERROR) # ③ 错误令牌 → 401（默认不强制也校验） st3, _h3, _b3 = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.9", "X-App-Token": "bad"}) self.assertEqual(st3, 401) # ④ 不带令牌（enforce=false）→ 照常可用（老 App） st4, _h4, _b4 = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.9"}) self.assertEqual(st4, 200) def test_enforced_over_the_wire(self): self.cfg.update({"app_token_enforce": True}) st, _h, body = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.9"}) self.assertEqual(st, 401, body) tok = self.token_for("203.0.113.9") st2, _h2, body2 = self.get(f"/music/do.php?musicno={TID}", {"X-Forwarded-For": "203.0.113.9", "X-App-Token": tok}) self.assertEqual(st2, 200, body2) def test_proxy_split_over_the_wire(self): """伪造 `X-Forwarded-Host`（127.0.0.1 可信）：媒体站能下文件+Range，其余 404。""" media_hdrs = {"X-Forwarded-For": "203.0.113.9", "X-Forwarded-Host": "md.kasugano.cn:4433"} st, hdrs, body = self.get(f"/media/{TID}.ts", media_hdrs) self.assertEqual(st, 200, body) self.assertEqual(hdrs.get("accept-ranges"), "bytes") self.assertEqual(len(body), 300) st2, hdrs2, body2 = self.get(f"/media/{TID}.ts", {**media_hdrs, "Range": "bytes=100-199"}) self.assertEqual(st2, 206, body2) self.assertEqual(hdrs2.get("content-range"), "bytes 100-199/300") self.assertEqual(len(body2), 100) # 同一宿主调 do.php → 404（媒体站只下文件） st3, _h3, _b3 = self.get(f"/music/do.php?musicno={TID}", media_hdrs) self.assertEqual(st3, 404) # API 宿主拉媒体 → 404 st4, _h4, _b4 = self.get(f"/media/{TID}.ts", {"X-Forwarded-For": "203.0.113.9", "X-Forwarded-Host": "mdmuse.kasugano.cn"}) self.assertEqual(st4, 404) # 不可信对端伪造宿主 → 只按真实 Host 判定 → 媒体站域名不成立 → 404 self.cfg.update({"trusted_proxies": ["192.168.100.10"]}) st5, _h5, _b5 = self.get(f"/media/{TID}.ts", media_hdrs) self.assertEqual(st5, 404) # ══════════════════════ ⑥ IIS 交付物静态校验 ══════════════════════ class TestIisSplitConfigs(unittest.TestCase): def read(self, path): with open(path, encoding="utf-8") as fh: return fh.read() def test_both_site_configs_parse_and_are_disjoint(self): for path in (API_CFG, MEDIA_CFG): self.assertTrue(os.path.isfile(path), f"缺少 {path}") text = self.read(path) root = ET.fromstring(text) # 必须能被 ElementTree 解析 self.assertTrue(root.tag.endswith("configuration")) self.assertNotIn("6678", text, "管理端口不该出现在任何对外配置里") self.assertIn("192.168.100.232:6677", text) # 反代到服务端口 6677 # IIS 里实际生效的是"服务器变量名"（HTTP_ 前缀那种写法），四个都要传 for name in ("HTTP_X_FORWARDED_FOR", "HTTP_X_REAL_IP", "HTTP_X_FORWARDED_PROTO", "HTTP_X_FORWARDED_HOST"): self.assertIn(name, text, f"{path} 缺少 {name}") self.assertIn("doDynamicCompression=\"false\"", text) # 关压缩（保 Range） rules = root.findall(".//{*}rule") self.assertTrue(rules, f"{path} 没有任何规则") self.assertEqual(rules[-1].find(".//{*}action").get("type"), "CustomResponse") self.assertEqual(rules[-1].find(".//{*}action").get("statusCode"), "404") def test_api_site_has_no_media_rule(self): text = self.read(API_CFG) self.assertNotIn("media", text.lower(), "API 站里不该出现任何媒体路径") for needle in ("token", "api.php", "i.php", "music/do.php", "push_download", "push_download/status", "healthz"): self.assertIn(needle, text, f"API 站缺少 {needle} 规则") # 反代规则数：7 条接口规则 + 1 条兜底 404 rules = ET.fromstring(text).findall(".//{*}rule") self.assertEqual(len(rules), 8, [r.get("name") for r in rules]) self.assertIn("api/client_ip", text) # 排查接口：默认注释，但要有 def test_media_site_is_media_only(self): text = self.read(MEDIA_CFG) self.assertIn("media", text) self.assertNotIn("do.php", text, "媒体站不该出现取址接口") self.assertNotIn("push_download", text, "媒体站不该出现推送接口") self.assertIn("4433", text) self.assertIn("preserveHostHeader", text) # ARR 保留原始 Host rules = ET.fromstring(text).findall(".//{*}rule") self.assertEqual(len(rules), 2, [r.get("name") for r in rules]) self.assertEqual(rules[0].find(".//{*}match").get("url"), "^(media/.*)$") # 目标必须是 `…:6677/media/<文件>`（用 {R:0} 时必须**没有**第二个 media 段） action = rules[0].find(".//{*}action").get("url") self.assertEqual(action, "http://192.168.100.232:6677/{R:0}") self.assertNotIn("/media/{R:0}", action) # ══════════════════════ ⑦ 管理界面契约（源码级） ══════════════════════ class TestUiContract(unittest.TestCase): """界面必须能看到/操作到这些开关（否则后端做了用户也不知道）。""" def setUp(self): from ktv import ui self.html = ui.PAGE def test_settings_expose_token_and_host_keys(self): for key in ("app_token_ttl_sec", "app_token_paths", "token_rate_limit_per_min", "media_hosts", "api_hosts"): self.assertIn(f"'{key}'", self.html, f"设置页缺少 {key}") for sel in ("s_app_token_bind_ip", "s_app_token_single_use", "s_app_token_persist", "s_app_token_enforce"): self.assertIn(f'id="{sel}"', self.html, f"设置页缺少 {sel}") # 说明里写明"为什么默认不强制"（老 App 红线）与端口容错 self.assertIn("X-App-Token", self.html) self.assertIn("老 App 不带令牌照常可用", self.html) self.assertIn("4433", self.html) if __name__ == "__main__": unittest.main(verbosity=2)