""" 配置模块 (config) 模块: 客户端 / 配置 输入: config.json 输出: 各模块配置注入 (crypto 拿密钥, transfer 拿地址) 配置外置, 源码不硬编码; 必填项缺失直接报错, 不带着坏配置跑。 默认配置文件: config/config.json (可 --init 生成模板) """ import argparse import json import os import sys from pathlib import Path from typing import Any, cast def _default_config() -> Path: """配置文件默认路径: SZ_CONFIG env -> XDG (~/.config/7z-encrypt/config.json) -> 兼容源码目录 config/config.json (开发模式) 安装版 (deb) 源码目录不可写, 配置走用户级 XDG。 """ env = os.environ.get("SZ_CONFIG") if env: return Path(env).expanduser() xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config") xdg_cfg = Path(xdg) / "7z-encrypt" / "config.json" if xdg_cfg.exists(): return xdg_cfg legacy = Path(__file__).resolve().parent / "config" / "config.json" return xdg_cfg if not legacy.exists() else legacy DEFAULT_CONFIG = _default_config() # 默认值 (用户 config.json 逐层覆盖) DEFAULTS: dict[str, Any] = { "server": {"url": "", "token": "", "timeout": 30, "max_retries": 3, "retry_delay": 1.0}, "crypto": {"key_id": "default_key", "keyring_path": ""}, "splitter": {"chunk_size_mb": 10}, "state": {"db_path": "~/.local/share/7z-encrypt/tasks.db", "cleanup_days": 7}, } # 必填路径 (缺失即报错) REQUIRED: list[tuple[str, str]] = [("server", "url")] class ConfigError(ValueError): """配置缺失或非法""" class AppConfig: """加载/校验/分发配置 (默认值兜底 + 用户覆盖)""" def __init__(self, config_path: str | Path | None = None) -> None: self.config_path = Path(config_path) if config_path else DEFAULT_CONFIG self.data: dict[str, Any] = {} self.load() def load(self) -> None: """读取 config.json, 与默认值合并, 校验必填项""" self.data = json.loads(json.dumps(DEFAULTS)) # 深拷贝默认值 if self.config_path.exists(): user = json.loads(self.config_path.read_text(encoding="utf-8")) self._merge(self.data, user) self.validate() @staticmethod def _merge(base: dict[str, Any], override: dict[str, Any]) -> None: for key, val in override.items(): if key in base and isinstance(base[key], dict) and isinstance(val, dict): # isinstance 收窄后是 dict[Unknown,Unknown], cast 回明确类型再递归 AppConfig._merge( cast(dict[str, Any], base[key]), cast(dict[str, Any], val) ) else: base[key] = val def validate(self) -> None: """必填校验 (服务器/密钥/卷大小), 任一非法抛 ConfigError""" for section, key in REQUIRED: v = self.data.get(section, {}).get(key) if not isinstance(v, str) or not v: raise ConfigError( f"必填配置缺失: {section}.{key} (用 --init 生成 config.json 模板)" ) url: str = self.data["server"]["url"] if not url.startswith(("http://", "https://")): raise ConfigError(f"server.url 必须以 http:// 或 https:// 开头: {url}") if self.data["splitter"]["chunk_size_mb"] <= 0: raise ConfigError("splitter.chunk_size_mb 必须大于 0") if self.data["server"]["max_retries"] < 1: raise ConfigError("server.max_retries 必须 >= 1") if self.data["server"]["timeout"] <= 0: raise ConfigError("server.timeout 必须大于 0") # ---------- 便捷访问 ---------- @property def server_url(self) -> str: return self.data["server"]["url"] @property def timeout(self) -> int: return self.data["server"]["timeout"] @property def max_retries(self) -> int: return self.data["server"]["max_retries"] @property def retry_delay(self) -> float: return self.data["server"]["retry_delay"] @property def token(self) -> str: """服务端认证 token (Bearer)""" return self.data["server"]["token"] @property def key_id(self) -> str: return self.data["crypto"]["key_id"] @property def keyring_path(self) -> Path | None: """密钥库路径: 空 = 系统默认 (SZ_KEYRING env / XDG 数据目录)""" raw = self.data["crypto"]["keyring_path"] return Path(raw) if raw else None @property def chunk_size(self) -> int: """卷大小 (字节)""" return self.data["splitter"]["chunk_size_mb"] * 1024 * 1024 @property def db_path(self) -> Path: return Path(self.data["state"]["db_path"]).expanduser() @property def cleanup_days(self) -> int: return self.data["state"]["cleanup_days"] # ---------- 分发: 构造各模块实例 ---------- def build_crypto(self): """crypto 拿密钥 (keyring 路径, None = 系统默认)""" from crypto import CryptoEngine return CryptoEngine(self.keyring_path) def build_transfer(self): """transfer 拿服务器地址/重试参数""" from transfer import TransferClient return TransferClient( self.server_url, timeout=self.timeout, max_retries=self.max_retries, retry_delay=self.retry_delay, ) def build_store(self): """state 拿数据库路径""" from state import TaskStore return TaskStore(self.db_path) def main() -> int: ap = argparse.ArgumentParser(description="配置: 加载校验 / 生成模板") ap.add_argument("--init", action="store_true", help="生成默认 config.json 模板") ap.add_argument("--path", default=str(DEFAULT_CONFIG), help="配置文件路径") ap.add_argument("--set-server", metavar="URL", help="更新 server.url 并保存") ap.add_argument("--set-token", metavar="TOKEN", help="更新 server.token 并保存") ap.add_argument("--regenerate-key", action="store_true", help="重新生成密钥 (红色警告 + 二次确认, 旧密文将无法解密)") ap.add_argument("--yes", action="store_true", help="跳过交互确认 (TUI 已确认时用, 需配 --regenerate-key)") args = ap.parse_args() if args.regenerate_key: cfg = AppConfig(args.path) cfg.load() engine = cfg.build_crypto() if not args.yes: print() print("\033[91m" + "═" * 52 + "\033[0m") print("\033[91m ⚠ 危险操作警告\033[0m") print() print("\033[91m 重新生成密钥 = 丢弃当前密钥!\033[0m") print("\033[91m · 当前密钥加密的已上传文件将永久无法解密\033[0m") print("\033[91m · 建议先导出旧密钥备份 (TUI 菜单 7)\033[0m") print("\033[91m · 执行后仅新上传的文件可用新密钥解密\033[0m") print("\033[91m" + "═" * 52 + "\033[0m") print() c1 = input("是否了解风险并继续? (输入 yes 继续): ").strip().lower() if c1 != "yes": print("[取消] 未重新生成") return 0 c2 = input("再次确认: 输入 CONFIRM 才执行: ").strip() if c2 != "CONFIRM": print("[取消] 未重新生成") return 0 engine.rotate_key() print(f"[完成] 密钥已重新生成 (keyring: {engine.keyring_path})") print("[提示] 旧密文已不可解; 需保留旧数据请从备份恢复旧密钥") return 0 if args.init: path = Path(args.path) if path.exists(): print(f"[错误] 已存在: {path}, 不覆盖") return 1 path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( { "server": { "url": "http://127.0.0.1:8000", "token": "", "timeout": 30, "max_retries": 3, "retry_delay": 1.0, }, "crypto": {"key_id": "default_key", "keyring_path": ""}, "splitter": {"chunk_size_mb": 10}, "state": {"db_path": "~/.local/share/7z-encrypt/tasks.db", "cleanup_days": 7}, }, indent=2, ensure_ascii=False, ) + "\n", encoding="utf-8", ) print(f"[config] 模板已生成: {path} (改 server.url 后即可用)") return 0 if args.set_server: path = Path(args.path) if not path.exists(): print(f"[错误] 配置文件不存在: {path}, 先 --init 生成") return 1 data = json.loads(path.read_text(encoding="utf-8")) data.setdefault("server", {})["url"] = args.set_server path.write_text( json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) print(f"[config] server.url -> {args.set_server} ({path})") return 0 if args.set_token: path = Path(args.path) if not path.exists(): print(f"[错误] 配置文件不存在: {path}, 先 --init 生成") return 1 data = json.loads(path.read_text(encoding="utf-8")) data.setdefault("server", {})["token"] = args.set_token path.write_text( json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) print(f"[config] server.token 已更新 ({path})") return 0 try: cfg = AppConfig(args.path) except (ConfigError, json.JSONDecodeError) as e: print(f"[错误] 配置无效: {e}") return 1 print(f"[config] 已加载: {cfg.config_path}") print(f" server: {cfg.server_url} (token={'已设置' if cfg.token else '未设置'}, timeout={cfg.timeout}, retries={cfg.max_retries})") print(f" crypto: key_id={cfg.key_id}, keyring={cfg.keyring_path}") print(f" splitter: 卷大小 {cfg.data['splitter']['chunk_size_mb']} MB") print(f" state: db={cfg.db_path}, 清理保留 {cfg.cleanup_days} 天") return 0 if __name__ == "__main__": sys.exit(main())