lou
279 lines
10 KiB
Python
279 lines
10 KiB
Python
"""
|
|
配置模块 (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())
|