maidong-ktv/selfhost-source/server/tests/test_app_token.py

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