Harden local server (Host/Origin gate vs DNS-rebind, confine import/report paths), atomic state-file writes, isolate per-creator save failures, empty-data + YouTube danmaku display guards

Author: HarryXin0919

scripts/app.py

@@ -214,8 +214,23 @@ def slug(s):
     return s or "imported"
 
 
+def _inside(child: Path, base: Path) -> bool:
+    """child 解析后是否落在 base 之内(解析符号链接,防越权读盘外文件)。"""
+    try:
+        c, b = child.resolve(), base.resolve()
+    except Exception:
+        return False
+    try:
+        return c.is_relative_to(b)            # Py3.9+
+    except AttributeError:
+        return c == b or b in c.parents       # 老 Python 兜底
+
+
 def do_import(path):
     p = Path(path.strip().strip('"').strip("'"))
+    # 只允许导入 DATA 目录内的文件:防攻击者借 /api/import 读任意 .json 再经 /api/data 外泄
+    if not _inside(p, DATA):
+        return {"ok": False, "error": "只能导入数据目录内的文件"}
     if not p.exists():
         return {"ok": False, "error": f"找不到文件:{p}"}
     if p.suffix.lower() != ".json":
@@ -243,6 +258,21 @@ class Handler(BaseHTTPRequestHandler):
     def log_message(self, *a):
         pass  # 安静,别刷屏
 
+    def _local_ok(self):
+        """防 DNS rebinding:只认本机 Host(精确到端口);POST 再校验 Origin 必须是 loopback。
+        远程网页拿到的 Host 是攻击者域名,据此拒掉,远程 JS 就驱动不了本地 API。"""
+        port = self.server.server_address[1]
+        host = (self.headers.get("Host") or "").strip()
+        if host not in (f"127.0.0.1:{port}", f"localhost:{port}"):
+            return False
+        if self.command == "POST":
+            origin = (self.headers.get("Origin") or "").strip()
+            if origin:
+                netloc = urlparse(origin).hostname
+                if netloc not in ("127.0.0.1", "localhost"):
+                    return False
+        return True
+
     def _json(self, code, obj):
         body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
         self.send_response(code)
@@ -263,6 +293,8 @@ def _file(self, target: Path):
         self.wfile.write(data)
 
     def do_GET(self):
+        if not self._local_ok():
+            return self._json(403, {"error": "forbidden"})
         path = urlparse(self.path).path
         if path in ("/", "/index.html"):
             if not GUI.exists():
@@ -337,12 +369,15 @@ def do_GET(self):
         if path == "/report" or path.startswith("/report/"):
             rel = path[len("/report"):].lstrip("/") or "index.html"
             target = (REPORTS / rel).resolve()
-            if not str(target).startswith(str(REPORTS.resolve())) or not target.exists():
+            # 真·目录包含(resolve 已归一化 ..);startswith 会被 reports_x 这类兄弟目录绕过
+            if not _inside(target, REPORTS) or not target.exists():
                 return self._json(404, {"error": "还没有报告 —— 先点『抓取并分析』生成"})
             return self._file(target)
         return self._json(404, {"error": "not found"})
 
     def do_POST(self):
+        if not self._local_ok():
+            return self._json(403, {"error": "forbidden"})
         path = urlparse(self.path).path
         length = int(self.headers.get("Content-Length", 0) or 0)
         raw = self.rfile.read(length) if length else b""

scripts/build_report.py

@@ -202,7 +202,7 @@ def section_table(videos):
             f'<td class="t">{title_html}</td>'
             f'<td data-v="{v.get("play") or 0}">{fmt_play(v.get("play"))}</td>'
             f'<td data-v="{v.get("comment") or 0}">{esc(v.get("comment"))}</td>'
-            f'<td data-v="{v.get("danmaku") or v.get("like") or 0}">{esc(v.get("danmaku") if v.get("danmaku") is not None else v.get("like"))}</td>'
+            f'<td data-v="{v.get("danmaku") or v.get("like") or 0}">{esc(v.get("danmaku") or v.get("like") or 0)}</td>'
             f'<td data-v="{v.get("duration_sec") or 0}">{fmt_dur(v.get("duration_sec"))}</td>'
             f'<td>{esc((v.get("created_iso") or "")[:10])}</td>'
             f'</tr>'

scripts/creator_profile.py

@@ -43,6 +43,16 @@ def safe_div(a, b):
 
 def creator_stats(vids):
     """一位 UP 主的结构性画像(可跨人对比的量)。"""
+    if not vids:
+        # 空数据集:返回零化但键齐全的画像,别让 len(feats)/len(vids) 除零崩掉整条 --report
+        return {
+            "play_med": 0,
+            "in_format_rate": 0,
+            "dur_med": 0,
+            "gap_days": None,
+            "comment_per_10k": 0,
+            "danmaku_per_min": 0,
+        }
     feats = [extract(v, NOW) for v in vids]
     plays = [f["play"] for f in feats]
     durs = [f["duration_sec"] for f in feats if f["duration_sec"] > 0]

scripts/enrich_bilibili.py

@@ -21,7 +21,7 @@
 
 from bilibili_api import video, Credential
 
-from runtime import DATA
+from runtime import DATA, atomic_write_text
 from creators import CREATORS
 
 try:
@@ -94,8 +94,8 @@ async def main():
                 await asyncio.sleep(PACE)
             n_ok += await enrich_one(v, cred)
             if n_ok and n_ok % 10 == 0:                  # 边取边存,中断不丢
-                p.write_text(json.dumps(vids, ensure_ascii=False, indent=2), encoding="utf-8")
-        p.write_text(json.dumps(vids, ensure_ascii=False, indent=2), encoding="utf-8")
+                atomic_write_text(p, json.dumps(vids, ensure_ascii=False, indent=2))
+        atomic_write_text(p, json.dumps(vids, ensure_ascii=False, indent=2))
         print(f"  ✓ {c['name']:<16} 三连补全 {n_ok}/{len(todo)} 条")
 
     print("\n✅ 完成。重跑 scan_signals.py / 报告,会自动出现「三连率 / 投币率」维度")

scripts/fetch_covers.py

@@ -30,7 +30,7 @@
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
-from runtime import DATA           # 源码=仓库/data,打包成 app 时=用户数据目录
+from runtime import DATA, atomic_write_text   # 源码=仓库/data,打包成 app 时=用户数据目录
 from creators import CREATORS
 
 HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
@@ -91,7 +91,10 @@ def main():
             print(f"  ✗ 缺 {c['alias']}_videos.json"); continue
         vids = json.loads(vp.read_text(encoding="utf-8"))
         cp = DATA / f"{c['alias']}_covers.json"
-        cache = json.loads(cp.read_text(encoding="utf-8")) if cp.exists() else {}
+        try:
+            cache = json.loads(cp.read_text(encoding="utf-8")) if cp.exists() else {}
+        except Exception:
+            cache = {}   # 上次写了一半损坏了:当没缓存重建,别让整轮崩在读取上
 
         jobs = [(v.get("bvid"), v.get("cover_url")) for v in vids
                 if v.get("bvid") and v.get("cover_url") and v.get("bvid") not in cache]
@@ -116,8 +119,8 @@ def work(bvid, url):
                     cache[bvid] = m
                     n_new += 1
                     if n_new % 10 == 0:                    # 边下边存,中断不丢
-                        cp.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8")
-        cp.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8")
+                        atomic_write_text(cp, json.dumps(cache, ensure_ascii=False, indent=2))
+        atomic_write_text(cp, json.dumps(cache, ensure_ascii=False, indent=2))
         print(f"  ✓ {c['name']}: 封面 {len(cache)}/{len(vids)} (新增 {n_new}, 失败 {n_fail})")
 
     print("\n✅ 封面指标已缓存,可重跑 scan_signals.py 看封面与播放的相关性")

scripts/fetch_multi.py

@@ -26,7 +26,7 @@
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
-from runtime import DATA           # 可写数据目录(源码=仓库/data,app=用户目录;runtime 已自动建好)
+from runtime import DATA, atomic_write_text   # 可写数据目录(源码=仓库/data,app=用户目录;runtime 已自动建好)
 
 try:
     from config_local import SESSDATA
@@ -65,7 +65,7 @@ def _save_result(c, platform, videos):
         print(f"  ✗ {c['name']} [{platform}]: 没抓到视频")
         return False
     out = DATA / f"{c['alias']}_videos.json"
-    out.write_text(json.dumps(videos, ensure_ascii=False, indent=2), encoding="utf-8")
+    atomic_write_text(out, json.dumps(videos, ensure_ascii=False, indent=2))
     hi, lo = videos[0], videos[-1]               # 适配器按播放降序返回:hi=最高播放,lo=最低
     # 时间跨度要按日期算,不能拿播放排序的首尾凑(那是"最低播放的日期 ~ 最高播放的日期")
     isos = sorted(i for i in (v.get("created_iso") for v in videos) if i)
@@ -111,10 +111,10 @@ async def run_bili_serial():
                 await asyncio.sleep(2.0)
             try:
                 videos = await fetch_for(c)
+                n_ok += _save_result(c, "bilibili", videos)   # 写入也包进 try:一人写失败不拖垮已抓到的别人
             except Exception as e:
                 print(f"  ✗ {c['name']} [bilibili]: {type(e).__name__}: {e}")
                 continue
-            n_ok += _save_result(c, "bilibili", videos)
 
     async def run_others_parallel():
         """YouTube(及未来的无风控平台):线程池并发;按清单顺序收割,输出顺序稳定。"""
@@ -134,13 +134,13 @@ def work(c, platform):
             for (c, platform), t in zip(others, tasks):
                 try:
                     videos = await t
+                    n_ok += _save_result(c, platform, videos)   # 写入也包进 try:一人写失败不拖垮已抓到的别人
                 except Exception as e:
                     print(f"  ✗ {c['name']} [{platform}]: {type(e).__name__}: {e}")
                     continue
-                n_ok += _save_result(c, platform, videos)
 
     # 两条线同时跑:B 站在等风控间隔时,YouTube 在并行抓
-    await asyncio.gather(run_bili_serial(), run_others_parallel())
+    await asyncio.gather(run_bili_serial(), run_others_parallel(), return_exceptions=True)
     print(f"\n✅ 完成,{n_ok} 个创作者已写 data/<alias>_videos.json")
 
 

scripts/runtime.py

@@ -79,6 +79,16 @@ def bootstrap() -> None:
     _BOOTSTRAPPED = True
 
 
+def atomic_write_text(path, text: str, encoding: str = "utf-8") -> None:
+    """原子写文本:先写同目录临时文件,再 os.replace 覆盖目标。
+    中断(断电/Ctrl+C)时目标文件要么是旧内容、要么是新内容,绝不会是写了一半的残缺数据。
+    用于不可再生的主数据(<alias>_videos.json / 封面缓存等)。"""
+    path = Path(path)
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(text, encoding=encoding)
+    os.replace(tmp, path)
+
+
 def worker_cmd(script: str, args=None):
     """拼出「用本工具自己的解释器跑某个子脚本」的命令行。
       源码模式: [python, scripts/<script>, *args]            (和历史行为一致)