diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index e2d4d35..b8000ac 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -267,6 +267,7 @@ PRAGMA synchronous = NORMAL; 业务参数放在 `config.jsonc`(JSONC:`//` / `/* */` 注释与尾逗号;实现为自研轻量剥离后再 `json.loads`,无额外依赖)。 敏感凭据放同目录 `secrets.json`(纯 JSON,供 `auth` 原子回写)。二者都必须被 Git 忽略,日志中不得输出敏感值。 默认路径优先 `config.jsonc`,否则回落 `config.json`;旧单体 `config.json`(凭据内嵌)仍可加载,若存在 `secrets.json` 则凭据以它为准。 +从旧单体迁到拆分布局:`uv run python tools/split_config_secrets.py [--dry-run] [--force] [--keep-src]`(默认写出同目录 `config.jsonc` + `secrets.json`,并把源改名为 `.bak`)。 `config.jsonc` 示例: diff --git a/README.md b/README.md index af26790..25e1301 100644 --- a/README.md +++ b/README.md @@ -314,6 +314,13 @@ Copy-Item secrets.example.json secrets.json 默认启动优先读 `config.jsonc`,若不存在则回落 `config.json`(旧单体文件仍可用)。 `python app.py auth` **只回写** `secrets.json`,不会冲掉 JSONC 注释。 +若部署目录仍是旧版单体 `config.json`,可用拆分脚本一次迁出: + +```powershell +uv run python tools/split_config_secrets.py --dry-run +uv run python tools/split_config_secrets.py # 写出 config.jsonc + secrets.json,源改名为 config.json.bak +``` + 凭据文件双账号格式: ```json diff --git a/tests/test_split_config_secrets.py b/tests/test_split_config_secrets.py new file mode 100644 index 0000000..7b0ceff --- /dev/null +++ b/tests/test_split_config_secrets.py @@ -0,0 +1,194 @@ +# -*- coding: utf-8 -*- +"""config.json → jsonc + secrets 拆分脚本测试(离线): +uv run python tests/test_split_config_secrets.py""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board import jsonc # noqa: E402 +from reaction_board.config import load_config # noqa: E402 +from tools.split_config_secrets import ( # noqa: E402 + split_config_file, + split_raw, +) + + +def _monolith() -> dict: + return { + "bilibili": { + "read": {"mid": 11, "csrf": "read-csrf", "cookie": "SESSDATA=secret-read"}, + "write": {"mid": 22, "csrf": "write-csrf", "cookie": "SESSDATA=secret-write"}, + }, + "telegram": { + "enabled": True, + "bot_token": "tg-secret-token", + "chat_id": 12345, + }, + "web": { + "host": "127.0.0.1", + "port": 8080, + "auth_token": "web-secret-token", + "public_base_url": "", + }, + "folders": [{"year": 2026, "media_id": 1, "title": "t"}], + "track_fans": [37754047], + "up_blacklist_seed": [], + } + + +def test_split_raw_extracts_secrets_keeps_business(): + business, secrets, stats = split_raw(_monolith()) + assert "bilibili" not in business + assert business["telegram"] == {"enabled": True} + assert business["web"]["host"] == "127.0.0.1" + assert business["web"]["port"] == 8080 + assert "auth_token" not in business["web"] + assert secrets["bilibili"]["read"]["mid"] == 11 + assert secrets["telegram"] == {"bot_token": "tg-secret-token", "chat_id": 12345} + assert secrets["web"] == {"auth_token": "web-secret-token"} + assert "bilibili" in stats["moved_paths"] + assert "telegram.bot_token" in stats["moved_paths"] + assert "web.auth_token" in stats["moved_paths"] + + +def test_split_raw_omits_absent_secret_stubs(): + raw = { + "telegram": {"enabled": False}, + "web": {"host": "0.0.0.0", "port": 9}, + "folders": [], + } + business, secrets, stats = split_raw(raw) + assert secrets == {} + assert business["telegram"] == {"enabled": False} + assert business["web"]["host"] == "0.0.0.0" + assert stats["moved_count"] == 0 + + +def test_dry_run_writes_nothing(): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + src = root / "config.json" + src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8") + summary = split_config_file(src, dry_run=True) + assert summary["dry_run"] is True + assert summary["moved_count"] >= 3 + assert not (root / "config.jsonc").exists() + assert not (root / "secrets.json").exists() + assert src.exists() + # 摘要不得泄露 cookie/token + blob = json.dumps(summary, ensure_ascii=False) + assert "SESSDATA" not in blob + assert "tg-secret-token" not in blob + assert "web-secret-token" not in blob + + +def test_refuses_overwrite_without_force(): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + src = root / "config.json" + src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8") + (root / "config.jsonc").write_text("// existing\n{}\n", encoding="utf-8") + try: + split_config_file(src, keep_src=True) + except FileExistsError as exc: + assert "config.jsonc" in str(exc) + assert "--force" in str(exc) + else: + raise AssertionError("已有输出时无 --force 必须失败") + assert src.exists() + assert "existing" in (root / "config.jsonc").read_text(encoding="utf-8") + + +def test_writes_and_renames_src_by_default(): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + src = root / "config.json" + src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8") + summary = split_config_file(src) + cfg_path = root / "config.jsonc" + sec_path = root / "secrets.json" + bak = root / "config.json.bak" + assert summary["ok"] is True + assert not src.exists() + assert bak.exists() + assert cfg_path.exists() + assert sec_path.exists() + + text = cfg_path.read_text(encoding="utf-8") + assert text.startswith("//") + assert "secrets.json" in text + business = jsonc.load(cfg_path) + secrets = json.loads(sec_path.read_text(encoding="utf-8")) + assert "bilibili" not in business + assert business["telegram"]["enabled"] is True + assert "bot_token" not in business["telegram"] + assert business["web"]["host"] == "127.0.0.1" + assert "auth_token" not in business["web"] + assert secrets["bilibili"]["read"]["mid"] == 11 + assert secrets["telegram"]["chat_id"] == 12345 + assert secrets["web"]["auth_token"] == "web-secret-token" + + loaded = load_config(cfg_path, env={}, require_auth=False) + assert loaded.bilibili.read.mid == 11 + assert loaded.bilibili.write.mid == 22 + assert loaded.telegram.enabled is True + assert loaded.telegram.bot_token == "tg-secret-token" + assert loaded.web.host == "127.0.0.1" + assert loaded.web.auth_token == "web-secret-token" + assert summary["verify"]["checked"] is True + assert summary["verify"]["read_mid"] == 11 + + +def test_force_overwrite_and_keep_src(): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + src = root / "config.json" + src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8") + (root / "config.jsonc").write_text("{}\n", encoding="utf-8") + (root / "secrets.json").write_text("{}\n", encoding="utf-8") + summary = split_config_file(src, force=True, keep_src=True) + assert src.exists() + assert summary["src_renamed"] is False + secrets = json.loads((root / "secrets.json").read_text(encoding="utf-8")) + assert secrets["bilibili"]["read"]["mid"] == 11 + + +def test_honors_secrets_path_relative(): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + src = root / "config.json" + data = _monolith() + data["secrets_path"] = "creds/my-secrets.json" + src.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + summary = split_config_file(src, keep_src=True) + sec = root / "creds" / "my-secrets.json" + assert Path(summary["secrets_out"]) == sec + assert sec.exists() + business = jsonc.load(root / "config.jsonc") + assert business.get("secrets_path") == "creds/my-secrets.json" + + +def main(): + test_split_raw_extracts_secrets_keeps_business() + test_split_raw_omits_absent_secret_stubs() + test_dry_run_writes_nothing() + test_refuses_overwrite_without_force() + test_writes_and_renames_src_by_default() + test_force_overwrite_and_keep_src() + test_honors_secrets_path_relative() + print( + "OK: split_config_secrets 抽取 bilibili/telegram/web 凭据、" + "dry-run、拒绝覆盖、改名与 secrets_path 测试全部通过" + ) + + +if __name__ == "__main__": + main() diff --git a/tools/split_config_secrets.py b/tools/split_config_secrets.py new file mode 100644 index 0000000..7adad0b --- /dev/null +++ b/tools/split_config_secrets.py @@ -0,0 +1,326 @@ +# -*- coding: utf-8 -*- +"""将旧版单体 ``config.json`` 拆成 ``config.jsonc`` + ``secrets.json``。 + +从源文件抽出凭据段: + +- 整段 ``bilibili``(若存在) +- ``telegram.bot_token`` / ``telegram.chat_id``(``enabled`` 等非密钥留在业务文件) +- ``web.auth_token``(``host`` / ``port`` / ``track`` 等留在业务文件) + +源里没有的密钥键不会写空壳。``telegram`` / ``web`` 抽空后若无剩余键,则不写入 +secrets;业务侧若还有非密钥键则保留对应段。``bilibili`` 整段移出业务文件。 + +用法:: + + # 先看会挪哪些键(不打印 cookie/token 原文) + uv run python tools/split_config_secrets.py --dry-run + + # 默认:写 config.jsonc + secrets.json,并把源文件改名为 config.json.bak + uv run python tools/split_config_secrets.py /path/to/config.json + + # 已有输出时需显式覆盖;保留源文件不改名 + uv run python tools/split_config_secrets.py --force --keep-src +""" + +from __future__ import annotations + +import argparse +import copy +import json +import os +import sys +from pathlib import Path +from typing import Any, Mapping, Sequence + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from reaction_board import jsonc # noqa: E402 +from reaction_board.config import load_config, resolve_secrets_path # noqa: E402 + +TELEGRAM_SECRET_KEYS = ("bot_token", "chat_id") +WEB_SECRET_KEYS = ("auth_token",) + +CONFIG_HEADER = """\ +// 业务配置(可含注释与尾逗号)。敏感凭据在同目录 secrets.json。 +// `python app.py auth` 只回写 secrets.json,不会冲掉本文件注释。 +// 本文件由 tools/split_config_secrets.py 从旧版单体配置拆出。 +""" + + +def _is_mapping(value: Any) -> bool: + return isinstance(value, Mapping) and not isinstance(value, (str, bytes)) + + +def split_raw(raw: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + """拆分业务配置与凭据。 + + 返回 ``(business, secrets, stats)``。``stats`` 仅含键路径计数,不含密钥值。 + """ + business: dict[str, Any] = copy.deepcopy(dict(raw)) + secrets: dict[str, Any] = {} + moved: list[str] = [] + + bilibili = business.pop("bilibili", None) + if _is_mapping(bilibili): + secrets["bilibili"] = copy.deepcopy(dict(bilibili)) + moved.append("bilibili") + elif bilibili is not None: + # 非法类型原样丢掉业务侧,避免残留;不写入 secrets + moved.append("bilibili(non-object-dropped)") + + telegram = business.get("telegram") + if _is_mapping(telegram): + tg_biz = dict(telegram) + tg_sec: dict[str, Any] = {} + for key in TELEGRAM_SECRET_KEYS: + if key in tg_biz: + tg_sec[key] = tg_biz.pop(key) + moved.append(f"telegram.{key}") + if tg_sec: + secrets["telegram"] = tg_sec + if tg_biz: + business["telegram"] = tg_biz + else: + business.pop("telegram", None) + + web = business.get("web") + if _is_mapping(web): + web_biz = dict(web) + web_sec: dict[str, Any] = {} + for key in WEB_SECRET_KEYS: + if key in web_biz: + web_sec[key] = web_biz.pop(key) + moved.append(f"web.{key}") + if web_sec: + secrets["web"] = web_sec + if web_biz: + business["web"] = web_biz + else: + business.pop("web", None) + + stats = { + "moved_paths": moved, + "moved_count": len(moved), + "business_top_keys": sorted(business.keys()), + "secrets_top_keys": sorted(secrets.keys()), + "has_bilibili": "bilibili" in secrets, + "telegram_secret_keys": sorted(secrets.get("telegram", {}).keys()) + if isinstance(secrets.get("telegram"), dict) + else [], + "web_secret_keys": sorted(secrets.get("web", {}).keys()) + if isinstance(secrets.get("web"), dict) + else [], + } + return business, secrets, stats + + +def _atomic_write_json(path: Path, data: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + with tmp.open("w", encoding="utf-8") as fh: + json.dump(data, fh, ensure_ascii=False, indent=2) + fh.write("\n") + os.replace(tmp, path) + + +def _atomic_write_jsonc(path: Path, data: Mapping[str, Any], *, header: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + body = json.dumps(data, ensure_ascii=False, indent=2) + if not body.endswith("\n"): + body += "\n" + text = header.rstrip() + "\n" + body + tmp = path.with_name(path.name + ".tmp") + with tmp.open("w", encoding="utf-8") as fh: + fh.write(text) + os.replace(tmp, path) + + +def resolve_outputs( + src: Path, + raw: Mapping[str, Any], + *, + config_out: Path | None, + secrets_out: Path | None, +) -> tuple[Path, Path]: + """解析输出路径;``secrets_path`` 相对源文件目录。""" + base = src.resolve().parent + cfg_path = config_out if config_out is not None else (base / "config.jsonc") + if secrets_out is not None: + sec_path = secrets_out + else: + sec_path = resolve_secrets_path(src, raw) + return cfg_path, sec_path + + +def split_config_file( + src: Path, + *, + config_out: Path | None = None, + secrets_out: Path | None = None, + dry_run: bool = False, + force: bool = False, + keep_src: bool = False, +) -> dict[str, Any]: + """执行拆分;返回可 JSON 序列化的摘要(不含密钥原文)。""" + src = Path(src) + if not src.exists(): + raise FileNotFoundError(f"源配置不存在: {src}") + + try: + raw = jsonc.load(src) + except ValueError as exc: + raise ValueError(f"源配置不是合法 JSON/JSONC: {src}: {exc}") from exc + if not isinstance(raw, Mapping): + raise ValueError(f"源配置顶层必须是对象: {src}") + + business, secrets, stats = split_raw(raw) + cfg_path, sec_path = resolve_outputs( + src, raw, config_out=config_out, secrets_out=secrets_out + ) + + summary: dict[str, Any] = { + "ok": True, + "dry_run": bool(dry_run), + "src": str(src), + "config_out": str(cfg_path), + "secrets_out": str(sec_path), + "keep_src": bool(keep_src), + "src_bak": None if keep_src else str(src.with_name(src.name + ".bak")), + **stats, + } + + if dry_run: + return summary + + conflicts = [p for p in (cfg_path, sec_path) if p.exists()] + if conflicts and not force: + names = ", ".join(str(p) for p in conflicts) + raise FileExistsError( + f"输出已存在(需 --force 覆盖): {names}" + ) + + _atomic_write_jsonc(cfg_path, business, header=CONFIG_HEADER) + _atomic_write_json(sec_path, secrets) + + # 可选校验:合并加载后 mid 与源一致(不打印 cookie) + verify: dict[str, Any] = {"checked": False} + if secrets.get("bilibili") or raw.get("bilibili"): + cfg = load_config(cfg_path, env={}, require_auth=False) + src_bili = raw.get("bilibili") if _is_mapping(raw.get("bilibili")) else {} + read_mid = None + write_mid = None + if _is_mapping(src_bili.get("read")): + read_mid = src_bili["read"].get("mid") + if _is_mapping(src_bili.get("write")): + write_mid = src_bili["write"].get("mid") + if read_mid is not None and int(cfg.bilibili.read.mid) != int(read_mid): + raise RuntimeError( + f"校验失败: bilibili.read.mid {cfg.bilibili.read.mid} != {read_mid}" + ) + if write_mid is not None and int(cfg.bilibili.write.mid) != int(write_mid): + raise RuntimeError( + f"校验失败: bilibili.write.mid {cfg.bilibili.write.mid} != {write_mid}" + ) + verify = { + "checked": True, + "read_mid": int(cfg.bilibili.read.mid), + "write_mid": int(cfg.bilibili.write.mid), + } + summary["verify"] = verify + + if not keep_src: + bak = src.with_name(src.name + ".bak") + if bak.exists() and not force: + raise FileExistsError( + f"备份已存在(需 --force 覆盖): {bak}" + ) + if bak.exists(): + bak.unlink() + os.replace(src, bak) + summary["src_bak"] = str(bak) + summary["src_renamed"] = True + else: + summary["src_renamed"] = False + + return summary + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="拆分旧版 config.json 为 config.jsonc + secrets.json", + ) + parser.add_argument( + "src_pos", + nargs="?", + default=None, + help="源配置路径(默认仓库根目录 config.json)", + ) + parser.add_argument( + "--src", + default=None, + help="源配置路径(与位置参数二选一;默认 ROOT/config.json)", + ) + parser.add_argument( + "--config-out", + default=None, + help="业务 JSONC 输出路径(默认与源同目录 config.jsonc)", + ) + parser.add_argument( + "--secrets-out", + default=None, + help="凭据 JSON 输出路径(默认同目录 secrets.json," + "或源内 secrets_path)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="只打印将挪动的键路径,不写文件、不改名", + ) + parser.add_argument( + "--force", + action="store_true", + help="允许覆盖已存在的输出(及 .bak)", + ) + parser.add_argument( + "--keep-src", + action="store_true", + help="成功写入后保留源文件;默认改名为 .bak," + "以便 default_config_path 优先读到新的 config.jsonc", + ) + args = parser.parse_args(list(argv) if argv is not None else None) + + if args.src_pos and args.src and args.src_pos != args.src: + print( + json.dumps( + {"ok": False, "error": "位置参数与 --src 冲突"}, + ensure_ascii=False, + ), + file=sys.stderr, + ) + return 2 + src = Path(args.src_pos or args.src or (ROOT / "config.json")) + config_out = Path(args.config_out) if args.config_out else None + secrets_out = Path(args.secrets_out) if args.secrets_out else None + + try: + summary = split_config_file( + src, + config_out=config_out, + secrets_out=secrets_out, + dry_run=args.dry_run, + force=args.force, + keep_src=args.keep_src, + ) + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return 0 + except (OSError, ValueError, RuntimeError) as exc: + print( + json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), + file=sys.stderr, + ) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main())