"""App 令牌（默认 10 分钟时效）与**双站点反代**的宿主隔离。

背景（为什么要有这一层）：
  * 已发布的官方 App 直接调 `GET /i.php` 与 `GET /music/do.php`，**不带**我们的令牌；
    所以令牌校验默认**只"带了就校验"**（`app_token_enforce=false`），绝不能强制 ——
    否则老 App 立刻全部 401（这是硬红线，见 `service.py` 顶部与 `deploy/iis/README.md`）。
  * 未来的自有 App 想用一个短时效随机串换地址：先 `GET /token` 拿令牌，再带
    `X-App-Token: <令牌>` 调 `/music/do.php` / `/push_download`。把 `app_token_enforce`
    打开即变成"必须带令牌"。
  * 反代拆两个站点（`mdmuse.kasugano.cn` 走 API、`md.kasugano.cn:4433` 只下文件）后，
    后端要能**只认宿主**把两类流量分开：媒体宿主不许调 API、API 宿主不许批量下文件。

三个接口（都在**服务端口** 6677）：
  * `GET/POST /token`        —— 签发一个随机令牌（`secrets.token_urlsafe(32)`）；
  * `GET /token/status`      —— 运维核对：累计签发/校验失败/限流计数 + 未汇总的当前分钟计数；
  * 其余受保护路径的校验发生在 `web.Router.guard`（**路由匹配之前**，见 `make_guard`）。

配置（都在 `ktv/config.py` 的 DEFAULTS 里，改完**立刻生效**、无需重启）：

  | 键 | 默认 | 说明 |
  |---|---|---|
  | `app_token_ttl_sec` | `600` | 有效期（秒），钳制 30~86400 |
  | `app_token_bind_ip` | `True` | 绑定**签发时的真实客户端 IP**（配合 `trusted_proxies`/`trust_proxy_headers` 取真实 IP）；换 IP 用同一令牌一律拒绝 |
  | `app_token_single_use` | `False` | 用过即失效（媒体地址是多次调用，**默认必须是 False**） |
  | `app_token_persist` | `False` | `True` = 令牌落 `meta` 表（重启后仍有效）；`False` = 纯内存（默认） |
  | `app_token_paths` | `"/music/do.php,/push_download"` | 受令牌管理的**路径前缀**列表（逗号分隔） |
  | `app_token_enforce` | `False` | `True` = 上面那些路径**必须**带令牌；`False` = 只"带了就校验" |
  | `token_rate_limit_per_min` | `60` | 同一真实客户端 IP 每分钟最多签发几个（0 = 关限流） |
  | `media_hosts` | `[]` | 媒体宿主（如 `["md.kasugano.cn"]`）；非空才启用宿主隔离 |
  | `api_hosts` | `[]` | API 宿主（如 `["mdmuse.kasugano.cn"]`） |

宿主隔离规则（`media_hosts` 为空 = 完全不校验，行为与改造前一致）：
  * `/media/*` 之外的路径若来自**媒体宿主** → 404（媒体站只能下文件）；
  * `/media/*` 若来自**非媒体宿主**且 `api_hosts` 非空 → 404（API 站不对外提供批量下载）；
  * 唯一豁免 `GET /api/client_ip`（排错入口，它会告诉你判定成了哪个宿主）。

安全前提：宿主判定只采信**经过 `ktv/proxy.py` 可信校验**的 `X-Forwarded-Host`，
不可信时退回 `Host` 头（见 `proxy.effective_host`）—— 否则谁都能自己塞一个
`X-Forwarded-Host: md.kasugano.cn:4433` 绕过隔离，或塞 API 宿主名去偷媒体。

日志口径（**这个项目很在意日志量**）：
  * 令牌**成功签发/成功校验不逐笔记**，只计数；
  * "签发 N 个"与"校验失败 N 次"按**自然分钟聚合成一条**，分钟翻页时写出（与
    `ktv/pushdl.py` 的重复/拒绝聚合同一个套路）；当前分钟还没汇总的计数在
    `GET /token/status` 的 `pending_log` 里也能看到，不会丢。

⚠️ 守卫（`make_guard`）里的任何异常都**不吞**：`web.py` 会统一回 500 并记日志。
   宁可响亮地失败，也不要在"强制令牌"打开时静默放行（fail-open 等于没有鉴权）。
"""
from __future__ import annotations

import hmac
import json
import re
import secrets
import threading
import time
from typing import Any, Callable

from . import proxy, web

# 默认受管路径（与 config.DEFAULTS["app_token_paths"] 同一份口径；消费方两种写法都认）
DEFAULT_PATHS = ("/music/do.php", "/push_download")
# 老 App 必走、且"带了令牌就校验"的入口（`/i.php` 默认不在 app_token_paths 里：
# `app_token_enforce=true` 时它**不强制**令牌，只有显式把它加进 app_token_paths 才强制）
LEGACY_CHECK_PATHS = ("/i.php",)
# `/i.php` 与 `/music/do.php` 的 `token` 参数是**老协议自己的**（= i.php 返回的 token），
# 不是我们的 App 令牌 —— 详见 `_is_legacy_param`。
LEGACY_PROTO_PATHS = ("/i.php", "/music/do.php")

TTL_MIN, TTL_MAX = 30, 86400
AUTH_ERROR = "令牌无效或已过期，请重新获取"
META_KEY = "app_tokens_v1"          # app_token_persist=true 时令牌落在 meta 表的这一行
MAX_IPS = 512                       # 限流窗口最多保留多少个来源 IP（防内存被撑爆）
MAX_TOKENS = 4096                   # 内存里最多保留多少个未过期令牌（防签发洪水撑爆内存）
_SAMPLE_MAX = 160
# cfg 挂不上属性时的兜底注册表：id(cfg) → (cfg, TokenService)（持有强引用，见 registry_for）
_FALLBACK: dict[int, tuple[Any, "TokenService"]] = {}


def normalize_paths(value: Any) -> list[str]:
    """把 `app_token_paths` 归一化成 `["/music/do.php", …]`。

    接受字符串（逗号/空白/分号分隔）或数组；缺前导 `/` 自动补上（配置层会给警告）。
    """
    if value is None:
        return []
    if isinstance(value, str):
        items = [x for x in re.split(r"[,\s;]+", value) if x]
    elif isinstance(value, (list, tuple, set)):
        items = [str(x) for x in value]
    else:
        items = [str(value)]
    out: list[str] = []
    for raw in items:
        p = str(raw).strip()
        if not p:
            continue
        out.append(p if p.startswith("/") else "/" + p)
    return out


def _int(value: Any, default: int) -> int:
    try:
        return int(value)
    except Exception:
        return default


def _bool(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    return str(value).strip().lower() in ("1", "true", "yes", "on")


def _no_store(resp: web.Response) -> web.Response:
    """响应**禁止任何缓存**（令牌是短时效凭证，被中间层缓存住会直接变成安全事故）。"""
    resp.headers["Cache-Control"] = "no-store"
    resp.headers["Pragma"] = "no-cache"
    return resp


def _not_found(req: web.Request) -> web.Response:
    """宿主隔离命中时的 404 —— 与"路径本来就不存在"**长得一模一样**（不透露隔离规则）。"""
    return web.json_response({"error": "not found", "path": req.path}, 404)


def unauthorized() -> web.Response:
    """令牌无效/过期/换 IP/没带（且已强制）时的统一 401 响应。"""
    return web.json_response({"ok": False, "error": AUTH_ERROR}, 401)


# ══════════════════════ 宿主隔离 ══════════════════════

def host_guard(cfg, req: web.Request) -> web.Response | None:
    """双站点宿主隔离：命中即 404，放行返回 None。

    `media_hosts` 为空（默认）= **完全不校验**（单站点部署行为不变）。

    ⚠️ 唯一豁免：`GET /api/client_ip`（排错入口）。理由：宿主配错时其它路径全是 404，
    若把排错口也 404 掉，就只能在服务端日志里猜"判定成了哪个宿主"；
    它自己会返回 `host` / `host_source` / `media_host_match` / `api_host_match`，一眼定位。
    它**仍然不建议对外暴露**（会返回内网代理地址列表）—— `deploy/iis/api-site/web.config`
    里这条规则默认是注释状态。
    """
    if str(req.path or "").startswith("/api/client_ip"):
        return None
    media = proxy.normalize_hosts(cfg.get("media_hosts"))
    if not media:
        return None
    api = proxy.normalize_hosts(cfg.get("api_hosts"))
    host, _src = proxy.effective_host(req)
    in_media_host = proxy.host_matches(host, media)
    path = str(req.path or "")
    if path.startswith("/media/"):
        # 媒体只能从媒体宿主下载；配了 api_hosts 时，其它宿主（含 API 站/未知宿主）一律 404
        if api and not in_media_host:
            return _not_found(req)
        return None
    if in_media_host:
        # 媒体宿主只许下文件：API、探活、推送、管理……统统 404
        return _not_found(req)
    return None


# ══════════════════════ 令牌服务 ══════════════════════

class TokenService:
    """`/token` 的签发/校验/限流/计数（一个实例挂在**同一份 cfg** 上，见 `registry_for`）。

    状态全在**内存**（进程内 dict）：令牌是 10 分钟量级的短时效凭证，重启后让 App 重新
    取一个即可；`app_token_persist=true` 时才把整份令牌表落到 `meta` 表（**一行**，
    不是"每个令牌写一次库"），具备重启后仍有效的能力。
    """

    def __init__(self, cfg, store=None, clock: Callable[[], float] | None = None):
        self.cfg = cfg
        self.store = store
        self._clock = clock or time.time       # 可注入时钟（单测用极小 TTL，不真等 10 分钟）
        self._lock = threading.RLock()
        self._tokens: dict[str, dict] = {}     # token -> {"exp": float, "ip": str}
        self._hits: dict[str, list[float]] = {}   # 来源 IP -> 最近一分钟内的签发时刻
        self._buckets: dict[int, dict] = {}    # 自然分钟 -> {"issued": n, "invalid": n, "sample": s}
        self.counters = {"issued": 0, "accepted": 0, "invalid": 0, "rate_limited": 0}
        self.last_at = 0
        self._loaded = False                   # 持久化令牌只加载一次

    # ── 配置（每次现读 → 改设置立刻生效）──
    def ttl(self) -> int:
        """有效期（秒），钳制 30~86400；填错按默认 600。"""
        v = _int(self.cfg.get("app_token_ttl_sec"), 600)
        return max(TTL_MIN, min(TTL_MAX, v))

    def bind_ip(self) -> bool:
        return _bool(self.cfg.get("app_token_bind_ip"), True)

    def single_use(self) -> bool:
        return _bool(self.cfg.get("app_token_single_use"), False)

    def persist(self) -> bool:
        return _bool(self.cfg.get("app_token_persist"), False)

    def enforce(self) -> bool:
        return _bool(self.cfg.get("app_token_enforce"), False)

    def paths(self) -> list[str]:
        raw = self.cfg.get("app_token_paths")
        return normalize_paths(DEFAULT_PATHS if raw is None else raw)

    def rate_per_min(self) -> int:
        """同一 IP 每分钟最多签发几个；`0` = 关闭限流。填错按默认 60。"""
        try:
            raw = self.cfg.get("token_rate_limit_per_min")
            return max(0, _int(60 if raw is None else raw, 60))
        except Exception:
            return 60

    # ── 持久化（app_token_persist=true 才用；默认全程内存，一次库都不写）──
    def _load_locked(self, now: float) -> None:
        if self._loaded:
            return
        self._loaded = True
        if not self.persist() or self.store is None:
            return
        try:
            raw = self.store.get_meta(META_KEY, "")
            data = json.loads(raw) if raw else {}
            toks = data.get("tokens") if isinstance(data, dict) else None
        except Exception:
            return
        if not isinstance(toks, dict):
            return
        for tok, rec in toks.items():
            try:
                exp = float(rec.get("exp") or 0)
            except Exception:
                continue
            if exp > now:
                self._tokens[str(tok)] = {"exp": exp, "ip": str(rec.get("ip") or "")}

    def _save_locked(self, now: float) -> None:
        if not self.persist() or self.store is None:
            return
        try:
            live = {t: {"exp": float(r["exp"]), "ip": str(r.get("ip") or "")}
                    for t, r in self._tokens.items() if float(r["exp"]) > now}
            self.store.set_meta(META_KEY, json.dumps({"tokens": live}, ensure_ascii=False))
        except Exception:
            pass                                # 落盘失败不影响内存里已经签发的令牌

    # ── 内存清理 ──
    def _prune_locked(self, now: float) -> int:
        """清掉过期令牌（顺带把超过上限的最老几个丢掉），返回清掉的数量。"""
        dead = [t for t, r in self._tokens.items() if float(r.get("exp") or 0) <= now]
        for t in dead:
            self._tokens.pop(t, None)
        if len(self._tokens) > MAX_TOKENS:
            order = sorted(self._tokens, key=lambda t: float(self._tokens[t].get("exp") or 0))
            for t in order[:len(self._tokens) - MAX_TOKENS]:
                self._tokens.pop(t, None)
                dead.append(t)
        return len(dead)

    # ── 签发 ──
    def issue(self, ip: str = "") -> dict:
        """签发一个令牌并返回 `{"ok","token","expires_in","expires_at"}`。"""
        now = self._clock()
        ttl = self.ttl()
        token = secrets.token_urlsafe(32)      # 43 字符 urlsafe（256 bit 熵）
        rec = {"exp": now + ttl, "ip": str(ip or "") if self.bind_ip() else ""}
        with self._lock:
            self._load_locked(now)
            self._prune_locked(now)            # 顺手清过期令牌（同一个锁内，避免并发写坏表）
            self._tokens[token] = rec
            self.counters["issued"] = int(self.counters.get("issued", 0)) + 1
            self.last_at = int(now)
            bucket = self._buckets.setdefault(int(now // 60), {})
            bucket["issued"] = int(bucket.get("issued", 0)) + 1
            self._save_locked(now)             # 只在**签发**（与清理）时落盘，不逐次校验写库
            lines = self._flush_locked(now)
        self._emit(lines)
        return {"ok": True, "token": token, "expires_in": ttl, "expires_at": int(rec["exp"])}

    # ── 校验 ──
    def _lookup_locked(self, token: str, ip: str, now: float) -> str:
        """返回空串 = 有效；否则是失败原因（`not_found` / `expired` / `ip_mismatch`）。"""
        rec = self._tokens.get(str(token or ""))
        if rec is None:
            return "not_found"
        if now >= float(rec.get("exp") or 0):
            return "expired"
        if self.bind_ip() and str(rec.get("ip") or "") and str(rec.get("ip")) != str(ip or ""):
            return "ip_mismatch"
        return ""

    def verify(self, token: str, ip: str = "") -> bool:
        """校验令牌（成功会记一次计数；失败按分钟聚合记一条 WARN 事件）。"""
        now = self._clock()
        tok = str(token or "").strip()
        with self._lock:
            self._load_locked(now)
            self._prune_locked(now)
            why = self._lookup_locked(tok, ip, now)
            if why == "":
                self.counters["accepted"] = int(self.counters.get("accepted", 0)) + 1
                self.last_at = int(now)
                if self.single_use():
                    self._tokens.pop(tok, None)
                    self._save_locked(now)
            else:
                self.counters["invalid"] = int(self.counters.get("invalid", 0)) + 1
                self.last_at = int(now)
                bucket = self._buckets.setdefault(int(now // 60), {})
                bucket["invalid"] = int(bucket.get("invalid", 0)) + 1
                if not bucket.get("sample"):
                    bucket["sample"] = f"来源IP={ip or '未知'} 原因={why}"[:_SAMPLE_MAX]
            lines = self._flush_locked(now)
        self._emit(lines)
        return why == ""

    # ── 限流（同一来源 IP 的滑动 60 秒窗口；与 ktv/pushdl.py 同一套路）──
    def rate_exceeded(self, ip: str) -> int:
        """返回还需等待的秒数（0 = 放行，并记一次签发时刻）。"""
        limit = self.rate_per_min()
        if limit <= 0:
            return 0
        now = self._clock()
        with self._lock:
            hits = [t for t in self._hits.get(ip, []) if now - t < 60.0]
            if len(hits) >= limit:
                self._hits[ip] = hits
                self.counters["rate_limited"] = int(self.counters.get("rate_limited", 0)) + 1
                bucket = self._buckets.setdefault(int(now // 60), {})
                bucket["rate_limited"] = int(bucket.get("rate_limited", 0)) + 1
                return max(1, int(60.0 - (now - hits[0])) + 1)
            hits.append(now)
            self._hits[ip] = hits
            if len(self._hits) > MAX_IPS:       # 淘汰最久没活动的来源
                overflow = len(self._hits) - MAX_IPS
                for k in sorted(self._hits, key=lambda x: self._hits[x][-1])[:overflow]:
                    self._hits.pop(k, None)
            return 0

    # ── 聚合日志（按自然分钟；成功签发/成功校验**不逐笔记**）──
    def _flush_locked(self, now: float) -> list[tuple[str, str]]:
        out: list[tuple[str, str]] = []
        cur = int(now // 60)
        for minute in sorted([m for m in self._buckets if m < cur]):
            b = self._buckets.pop(minute)
            if int(b.get("issued", 0)) > 0:
                out.append(("INFO", f"App 令牌：签发 {int(b['issued'])} 个"
                                    f"（每个有效期 {self.ttl()} 秒；不逐笔记录）"))
            if int(b.get("invalid", 0)) > 0 or int(b.get("rate_limited", 0)) > 0:
                tail = f"，示例：{b['sample']}" if b.get("sample") else ""
                out.append(("WARN", f"App 令牌：校验失败 {int(b.get('invalid', 0))} 次"
                                    f"（其中触发签发限流 {int(b.get('rate_limited', 0))} 次）{tail}"))
        return out

    def _emit(self, lines: list[tuple[str, str]]) -> None:
        for level, text in lines:
            try:
                self.store.event(level, text)
            except Exception:
                pass                            # store 不可用（单测传 None）时静默

    def status(self) -> dict:
        """运维核对：累计计数 + 生效口径 + **未汇总的当前分钟计数**。"""
        now = self._clock()
        with self._lock:
            lines = self._flush_locked(now)
            bucket = self._buckets.get(int(now // 60)) or {}
            pending = {k: int(bucket.get(k, 0)) for k in ("issued", "invalid", "rate_limited")
                       if int(bucket.get(k, 0)) > 0}
            if bucket.get("sample"):
                pending["sample"] = str(bucket["sample"])
            data = dict(self.counters)
            data.update({
                "ok": True, "last_at": int(self.last_at),
                "ttl_sec": self.ttl(), "rate_limit_per_min": self.rate_per_min(),
                "bind_ip": self.bind_ip(), "single_use": self.single_use(),
                "enforce": self.enforce(), "paths": self.paths(),
                "live_tokens": len(self._tokens), "pending_log": pending,
            })
        self._emit(lines)
        return data

    # ── 守卫（挂在 web.Router.guard 上，路由匹配之前跑）──
    def _in_paths(self, path: str) -> bool:
        return any(path.startswith(p) for p in self.paths())

    def _push_entry(self, path: str) -> bool:
        """`/push_download` 本体（**不含** `/push_download/status`）。"""
        return path in ("/push_download", "/push_download/")

    def _is_legacy_param(self, path: str, token: str) -> bool:
        """query/body 里的 `token` 是不是**老协议自己的令牌**（若是则不当 App 令牌看）。

        必须有这层区分，否则会踩两个坑：
          ① 老 App 的 `do.php?...&token=local`（= `/i.php` 返回的 token）会被当成
             "错误的 App 令牌" → 全部 401，直接破坏已发布 App（**红线**）；
          ② `/push_download` 的 body `token` 本来就是 `push_token`（ktv/pushdl.py 校验），
             当 App 令牌去校验会把点歌推送全部打成 401。
        自有 App 请统一用**请求头 `X-App-Token`**（那条通道没有任何歧义）。
        """
        if self._push_entry(path):
            return True
        if any(path.startswith(p) for p in LEGACY_PROTO_PATHS):
            legacy = str(self.cfg.get("service_token") or "local")
            return hmac.compare_digest(str(token or "").encode("utf-8"), legacy.encode("utf-8"))
        return False

    def _presented(self, req: web.Request) -> tuple[str, str]:
        """取出请求里"作为 App 令牌"的那一个值（返回 `(token, 来源)`）。

        `X-App-Token` 优先（明确通道）；其次才是 query/body 的 `token`
        （且必须不是老协议令牌，见 `_is_legacy_param`）。
        """
        header_tok = str(req.header("x-app-token") or "").strip()
        if header_tok:
            return header_tok, "x-app-token"
        path = str(req.path or "")
        param = str(req.query.get("token") or "").strip()
        if not param:
            try:
                param = str((req.form() or {}).get("token") or "").strip()
            except Exception:
                param = ""
        if param and not self._is_legacy_param(path, param):
            return param, "token"
        return "", ""

    def check_request(self, req: web.Request) -> web.Response | None:
        """令牌守卫：返回 401 响应 = 拒绝；None = 放行（并在放行时置 `req.app_token_ok`）。"""
        path = str(req.path or "")
        if not (self._in_paths(path) or any(path.startswith(p) for p in LEGACY_CHECK_PATHS)):
            return None                       # 不在受管范围：一个字节都不碰
        token, _src = self._presented(req)
        if token:
            if self.verify(token, str(req.client_ip or "")):
                req.app_token_ok = True
                return None
            return unauthorized()
        # 没带 App 令牌
        if self._push_entry(path):
            # `/push_download`：还有 `push_token` 那条合法通道，交给 ktv/pushdl.py 判
            # （它会给 401「鉴权失败」，口径保持不变）
            return None
        if self.enforce() and self._in_paths(path):
            return unauthorized()
        return None


# ══════════════════════ 注册表 / 路由 / 守卫装配 ══════════════════════

def registry_for(cfg, store=None) -> TokenService:
    """取（或建）**这份 cfg 对应**的令牌服务实例。

    ⚠️ 为什么挂在 cfg 上而不是模块级单例：单测在一个进程里会建很多份互不相干的
    Config（各自临时目录），模块级单例会把它们的令牌串在一起；挂 cfg 上天然隔离，
    且服务端口与管理端口拿到的是**同一个**实例（同一份 cfg）。

    极端情况下 cfg 不允许挂属性（`__slots__` 之类）：退回一张以 `id(cfg)` 为键、
    **同时持有 cfg 强引用**（防止 id 被复用）的兜底表 —— 关键是"签发"和"校验"必须
    拿到同一个实例，否则刚发的令牌永远校验不过（静默的严重故障）。
    """
    svc = getattr(cfg, "_app_token_service", None)
    if svc is None:
        svc = TokenService(cfg, store)
        try:
            setattr(cfg, "_app_token_service", svc)
        except Exception:                     # pragma: no cover - 只读对象兜底
            key = id(cfg)
            hit = _FALLBACK.get(key)
            if hit is None or hit[0] is not cfg:
                svc = TokenService(cfg, store)
                _FALLBACK[key] = (cfg, svc)
            else:
                svc = hit[1]
    elif store is not None and svc.store is None:
        svc.store = store                     # 先被 web 层建出来、后拿到 store 时补上
    return svc


def make_guard(cfg, store=None) -> Callable[[web.Request], web.Response | None]:
    """装配 `web.Router.guard`：先宿主隔离，再令牌校验。"""
    svc = registry_for(cfg, store)

    def guard(req: web.Request) -> web.Response | None:
        early = host_guard(cfg, req)
        if early is not None:
            return early
        return svc.check_request(req)

    return guard


def build_router(cfg, store=None) -> web.Router:
    """`GET/POST /token` + `GET /token/status`（挂在**服务端口**）。"""
    r = web.Router()
    svc = registry_for(cfg, store)

    def token_endpoint(req: web.Request):
        ip = str(req.client_ip or "") or "unknown"
        wait = svc.rate_exceeded(ip)
        if wait > 0:
            resp = web.json_response({
                "ok": False, "retry_after": wait, "client_ip": ip,
                "error": f"获取令牌过于频繁：来源 {ip} 每分钟最多 {svc.rate_per_min()} 个，"
                         f"请 {wait} 秒后再试"}, 429)
            resp.headers["Retry-After"] = str(wait)
            return _no_store(resp)
        # 令牌**绑定签发时的真实客户端 IP**（经可信代理时是设备 IP，见 ktv/proxy.py）
        return _no_store(web.json_response(svc.issue(str(req.client_ip or ""))))

    r.add("GET", "/token", token_endpoint)
    r.add("POST", "/token", token_endpoint)
    r.add("GET", "/token/status", lambda req: web.json_response(svc.status()))
    return r