zijunchiang
87aa979abd
版本号：
- 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 例）
521 lines
25 KiB
Python
521 lines
25 KiB
Python
"""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
|