diff --git a/README.md b/README.md index 63ae221..52ee5b7 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,30 @@ uv pip install --python .venv/bin/python nuitka TUI 自动优先调用编译产物（`bin/sz-transfer`），无则回退源码解释执行。 +## Termux / 手机安装 (deb 包) + +适配 PRoot Ubuntu / Termux 的安装包（配置走 XDG，`cwd` 无关）： + +```bash +# 构建 (需 debhelper): dpkg-buildpackage -us -uc -b +# 生成 ../7z-encrypt-client_1.0.0_all.deb + +# 手机上 (ZeroTermux PRoot Ubuntu) 安装: +apt install -y python3 python3-venv +dpkg -i 7z-encrypt-client_1.0.0_all.deb # postinst 自动建 venv + 装 cryptography/tqdm + +# 使用: +sz-config --init # 首次: 生成 ~/.config/7z-encrypt/config.json +sz-config --set-server http://服务器:8000 +sz-config --set-token 你的token +sz-transfer --list # CLI (服务器/token 自动读配置) +sz-tui # TUI (登录门禁, 中文菜单) + +# 换机密钥: 旧机导出 keyring -> 手机 ~/.local/share/7z-encrypt/keyring.json (权限 600) +``` + +依赖: `python3 >= 3.10` + `cryptography >= 50.0.0`（cobblestone 流式加密，postinst 自动装最新版）。 + ## 测试 ```bash diff --git a/config.py b/config.py index f160e04..08bcfec 100644 --- a/config.py +++ b/config.py @@ -10,18 +10,36 @@ import argparse import json +import os import sys from pathlib import Path from typing import Any, cast -DEFAULT_CONFIG = Path(__file__).resolve().parent / "config" / "config.json" +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": "config/tasks.db", "cleanup_days": 7}, + "state": {"db_path": "~/.local/share/7z-encrypt/tasks.db", "cleanup_days": 7}, } # 必填路径 (缺失即报错) @@ -117,7 +135,7 @@ class AppConfig: @property def db_path(self) -> Path: - return Path(self.data["state"]["db_path"]) + return Path(self.data["state"]["db_path"]).expanduser() @property def cleanup_days(self) -> int: @@ -204,7 +222,7 @@ def main() -> int: }, "crypto": {"key_id": "default_key", "keyring_path": ""}, "splitter": {"chunk_size_mb": 10}, - "state": {"db_path": "config/tasks.db", "cleanup_days": 7}, + "state": {"db_path": "~/.local/share/7z-encrypt/tasks.db", "cleanup_days": 7}, }, indent=2, ensure_ascii=False, diff --git a/config/keyring.json.bak b/config/keyring.json.bak deleted file mode 100644 index 05efec6..0000000 --- a/config/keyring.json.bak +++ /dev/null @@ -1,3 +0,0 @@ -{ - "default_key": "67e7c6baa066d932fc76fa59a76b8d3d7d0c782b8881cd9d0c4d93251565f8b2" -} \ No newline at end of file diff --git a/debian/.debhelper/generated/7z-encrypt-client/dh_installchangelogs.dch.trimmed b/debian/.debhelper/generated/7z-encrypt-client/dh_installchangelogs.dch.trimmed new file mode 100644 index 0000000..8cad0fd --- /dev/null +++ b/debian/.debhelper/generated/7z-encrypt-client/dh_installchangelogs.dch.trimmed @@ -0,0 +1,7 @@ +7z-encrypt-client (1.0.0) unstable; urgency=medium + + * 首个 Termux/PRoot 适配版: 配置 XDG 化 + venv 依赖自建 + * sz-transfer / sz-tui / sz-config 三个命令 + * 零知识 + 零合并 + 文件名加密 + 账号体系 + 登录门禁 + + -- edgevoid Mon, 10 Aug 2026 16:00:00 +0800 diff --git a/debian/.debhelper/generated/7z-encrypt-client/installed-by-dh_installdocs b/debian/.debhelper/generated/7z-encrypt-client/installed-by-dh_installdocs new file mode 100644 index 0000000..e69de29 diff --git a/debian/7z-encrypt-client.substvars b/debian/7z-encrypt-client.substvars new file mode 100644 index 0000000..978fc8b --- /dev/null +++ b/debian/7z-encrypt-client.substvars @@ -0,0 +1,2 @@ +misc:Depends= +misc:Pre-Depends= diff --git a/debian/7z-encrypt-client/DEBIAN/control b/debian/7z-encrypt-client/DEBIAN/control new file mode 100644 index 0000000..2b41e33 --- /dev/null +++ b/debian/7z-encrypt-client/DEBIAN/control @@ -0,0 +1,19 @@ +Package: 7z-encrypt-client +Version: 1.0.0 +Architecture: all +Maintainer: edgevoid +Installed-Size: 101 +Depends: python3 (>= 3.10), python3-venv +Section: utils +Priority: optional +Description: 端到端加密文件传输客户端 (Termux/PRoot 适配版) + 零知识 + 零合并的文件安全传输客户端: + 7z 压缩 -> cobblestone 加密 -> 分卷上传; 下载逐卷拉取本地解密。 + 服务端无密钥不解密, 文件名也加密 (enc: URL-safe base64)。 + . + 包含: + * sz-transfer CLI (上传/下载/列表/删除/配额/账号) + * sz-tui 终端界面 (登录门禁, 12 项菜单) + * sz-config 配置 (服务器/token/密钥) + . + 配置存 ~/.config/7z-encrypt/, 密钥存 ~/.local/share/7z-encrypt/。 diff --git a/debian/7z-encrypt-client/DEBIAN/md5sums b/debian/7z-encrypt-client/DEBIAN/md5sums new file mode 100644 index 0000000..e8a2e42 --- /dev/null +++ b/debian/7z-encrypt-client/DEBIAN/md5sums @@ -0,0 +1,12 @@ +440c53f84f5722b5146959c8fc6d7a23 usr/bin/sz-config +ec8cf0df841e479689ccb681f6168ccc usr/bin/sz-transfer +0ba5b8147990daa7fd07344952c42847 usr/bin/sz-tui +dcb31e74e0a6ee8073337fc4fb2acf99 usr/share/7z-encrypt/config.py +8a064c0b3c21d50812a3fd69a320d862 usr/share/7z-encrypt/crypto.py +21062e04f8c7f4471881a39c489dfa73 usr/share/7z-encrypt/metadata.py +3478ac8277b4b1a38694ad0cc8cdbcdb usr/share/7z-encrypt/splitter.py +b90b8c950f99e1439bc273f50a2ee0a6 usr/share/7z-encrypt/state.py +342458cd8e61337d5ee659fc24ccac8b usr/share/7z-encrypt/transfer.py +e60f9836c1eb125ac5b2296f9e906c8d usr/share/7z-encrypt/tui.py +198943ad7a94f6ba0995cf2d326f66d4 usr/share/doc/7z-encrypt-client/changelog.gz +fadf3972cc61c7c76d0ccc20b29a804c usr/share/doc/7z-encrypt-client/copyright diff --git a/debian/7z-encrypt-client/DEBIAN/postinst b/debian/7z-encrypt-client/DEBIAN/postinst new file mode 100755 index 0000000..1597b0a --- /dev/null +++ b/debian/7z-encrypt-client/DEBIAN/postinst @@ -0,0 +1,16 @@ +#!/bin/sh +set -e + +case "$1" in + configure) + P=/usr/share/7z-encrypt/.venv + if [ ! -x "$P/bin/python" ]; then + echo "正在创建 7z-encrypt 运行环境 (venv + cryptography/tqdm) ..." + python3 -m venv "$P" + "$P/bin/pip" install --quiet --upgrade pip || true + "$P/bin/pip" install --quiet cryptography tqdm + fi + ;; +esac + +exit 0 diff --git a/debian/7z-encrypt-client/DEBIAN/prerm b/debian/7z-encrypt-client/DEBIAN/prerm new file mode 100755 index 0000000..016f0f1 --- /dev/null +++ b/debian/7z-encrypt-client/DEBIAN/prerm @@ -0,0 +1,8 @@ +#!/bin/sh +set -e + +if [ "$1" = "remove" ] || [ "$1" = "purge" ]; then + rm -rf /usr/share/7z-encrypt/.venv 2>/dev/null || true +fi + +exit 0 diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/config.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/config.py new file mode 100644 index 0000000..08bcfec --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/config.py @@ -0,0 +1,278 @@ +""" +配置模块 (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()) diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/crypto.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/crypto.py new file mode 100644 index 0000000..a1566db --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/crypto.py @@ -0,0 +1,245 @@ +""" +加密引擎模块 (Crypto Engine) + +流式 AES-256-GCM 分块加密,基于 cryptography.cobblestone (C2SP chunked-encryption)。 +每块 16KiB 独立 GCM 认证,流式处理不整块进内存,适合大文件。 + +流程对应模块图 (02-加密引擎): + B(取密钥) -> C(初始化流式上下文) -> D(逐块加密) -> F(输出参数 + 密文流) + +格式说明: + cobblestone 输出流自带头部 + 分块结构 (magic/版本/随机 IV/密文+tag), + IV 由库内部管理,不需要也不应该手动维护 iv 列表。 + 解密时只需密钥 + context,finalize() 验证整体完整性,篡改即抛 InvalidTag。 +""" + +import base64 +import io +import json +import os +import shutil +from pathlib import Path +from typing import IO, Any, Callable + +from cryptography.cobblestone import Cobblestone256Decryptor, Cobblestone256Encryptor +from cryptography.exceptions import InvalidTag + +# 应用绑定上下文: 防止密文被挪用到其他系统/版本 +CONTEXT = b"7z-encrypt:v1" + +# 文件名加密独立 context (与文件内容加密隔离, 防交叉) +NAME_CTX = CONTEXT + b":name" + +# 默认密钥库位置 (系统级: 环境变量 SZ_KEYRING 优先, 否则 XDG 数据目录) +# 配置外置原则, 权限 600 +def _default_keyring() -> Path: + env = os.environ.get("SZ_KEYRING") + if env: + return Path(env).expanduser() + xdg = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share") + return Path(xdg) / "7z-encrypt" / "keyring.json" + + +DEFAULT_KEYRING = _default_keyring() + +# 旧位置 (v1: 项目目录 config/keyring.json) — 自动迁移到系统位置 +LEGACY_KEYRING = Path(__file__).resolve().parent / "config" / "keyring.json" + + +class CryptoError(RuntimeError): + """加密/解密过程错误""" + + +class KeyNotFoundError(CryptoError): + """密钥 ID 不存在""" + + +class IntegrityError(CryptoError): + """解密验证失败 (密钥错误或数据被篡改)""" + + +class CryptoEngine: + """流式加密引擎, 密钥持久化到 keyring 文件""" + + def __init__(self, keyring_path: str | Path | None = None) -> None: + self.keyring_path = Path(keyring_path) if keyring_path else DEFAULT_KEYRING + self._migrate_legacy_keyring() + self._keyring = self._load_keyring() + + def _migrate_legacy_keyring(self) -> None: + """旧位置密钥自动迁移到系统位置 (仅默认路径时, 避免密钥"丢失")""" + if self.keyring_path != DEFAULT_KEYRING: + return + if self.keyring_path.exists() or not LEGACY_KEYRING.exists(): + return + self.keyring_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(LEGACY_KEYRING, self.keyring_path) + os.chmod(self.keyring_path, 0o600) + print(f"[密钥库] 旧密钥已迁移 -> {self.keyring_path}") + + # ---------- 密钥库 ---------- + + def _load_keyring(self) -> dict[str, bytes]: + """从 keyring 文件加载密钥 (hex -> bytes)""" + if not self.keyring_path.exists(): + return {} + data = json.loads(self.keyring_path.read_text(encoding="utf-8")) + return {kid: bytes.fromhex(hex_str) for kid, hex_str in data.items()} + + def _save_keyring(self) -> None: + """持久化密钥库, 文件权限 600 (仅本人可读写)""" + self.keyring_path.parent.mkdir(parents=True, exist_ok=True) + data = {kid: key.hex() for kid, key in self._keyring.items()} + self.keyring_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + os.chmod(self.keyring_path, 0o600) + + def get_key(self, key_id: str) -> bytes: + """步骤 B: 取密钥。密钥不存在抛 KeyNotFoundError""" + try: + return self._keyring[key_id] + except KeyError: + raise KeyNotFoundError(f"密钥ID '{key_id}' 未找到 (keyring: {self.keyring_path})") from None + + def generate_key(self, key_id: str) -> bytes: + """生成新密钥 (os.urandom 32B) 并持久化。已存在则不覆盖, 抛错防误用""" + if key_id in self._keyring: + raise CryptoError(f"密钥ID '{key_id}' 已存在, 不覆盖 (需轮换请用 rotate_key)") + key = os.urandom(32) + self._keyring[key_id] = key + self._save_keyring() + return key + + def rotate_key(self, key_id: str = "default_key") -> bytes: + """强制轮换密钥: 删旧密钥 -> 生成新密钥。 + + 危险: 旧密钥加密的密文将永久无法解密, 调用方必须确认! + """ + self._keyring.pop(key_id, None) + return self.generate_key(key_id) + + # ---------- 文件名加密 (元数据零知识: 服务端看不到明文文件名) ---------- + + def encrypt_name(self, name: str, key_id: str = "default_key") -> str: + """文件名加密 -> 'enc:' + URL-safe base64 密文 (cobblestone 独立 context) + + URL-safe: 密文不含 / + =, 避免被服务端/路径逻辑误伤 + """ + key = self.get_key(key_id) + enc = Cobblestone256Encryptor(key, NAME_CTX) + out = io.BytesIO() + out.write(enc.update(name.encode())) + out.write(enc.finalize()) + return "enc:" + base64.urlsafe_b64encode(out.getvalue()).decode() + + def decrypt_name(self, token: str, key_id: str = "default_key") -> str: + """解密 'enc:' 前缀文件名, 失败抛 IntegrityError""" + raw = base64.urlsafe_b64decode(token[4:]) + key = self.get_key(key_id) + try: + dec = Cobblestone256Decryptor(key, NAME_CTX) + out = io.BytesIO() + out.write(dec.update(raw)) + out.write(dec.finalize()) + return out.getvalue().decode() + except InvalidTag as e: + raise IntegrityError("文件名解密失败: 密钥不匹配或数据损坏") from e + + # ---------- 加解密 ---------- + + def encrypt_stream( + self, + src: IO[bytes], + dst: IO[bytes], + key_id: str, + context: bytes = CONTEXT, + progress: Callable[[int], Any] | None = None, + ) -> dict[str, Any]: + """步骤 A & F: 流式加密。src 读明文 -> dst 写密文, 返回加密参数 + + Args: + src: 可读二进制流 (文件/BytesIO) + dst: 可写二进制流 + key_id: 密钥标识 + context: 应用绑定上下文 (默认 7z-encrypt:v1) + progress: 进度回调, 每读块后调 progress(已读字节) + """ + key = self.get_key(key_id) + enc = Cobblestone256Encryptor(key, context) + done = 0 + while True: + chunk = src.read(1 << 16) # 64KiB 读块 + if not chunk: + break + dst.write(enc.update(chunk)) + done += len(chunk) + if progress is not None: + progress(done) + dst.write(enc.finalize()) + return {"alg": "cobblestone-aes256gcm", "key_id": key_id, "context": context.decode()} + + def decrypt_stream( + self, + src: IO[bytes], + dst: IO[bytes], + key_id: str, + context: bytes = CONTEXT, + progress: Callable[[int], Any] | None = None, + ) -> None: + """解密流 (进度回调: 每读块后调 progress(已读字节))""" + key = self.get_key(key_id) + try: + dec = Cobblestone256Decryptor(key, context) + done = 0 + while True: + chunk = src.read(1 << 16) + if not chunk: + break + dst.write(dec.update(chunk)) # 块级认证: 篡改在这里就抛 InvalidTag + done += len(chunk) + if progress is not None: + progress(done) + dst.write(dec.finalize()) # finalize 返回剩余明文 + except InvalidTag as e: + raise IntegrityError("解密验证失败: 密钥不匹配或密文被篡改") from e + + +# ================= 使用示例 (Usage Example) ================= +if __name__ == "__main__": + import io + + engine = CryptoEngine() + + # 1. 确保密钥存在 (首次运行生成并持久化) + try: + engine.get_key("default_key") + except KeyNotFoundError: + engine.generate_key("default_key") + print("[密钥库] 已生成 default_key ->", engine.keyring_path) + + # 2. 模拟流式加密 (大文件场景: 分块读入) + plain = (b"This is a streaming encryption test. " * 10000) # ~330KB + src = io.BytesIO(plain) + dst = io.BytesIO() + params = engine.encrypt_stream(src, dst, key_id="default_key") + + print("=== 加密输出 (JSON) ===") + print(json.dumps(params, indent=2)) + print(f"明文 {len(plain)}B -> 密文 {len(dst.getvalue())}B (含头部+分块tag开销)") + + # 3. 流式解密还原 + out = io.BytesIO() + dst.seek(0) + engine.decrypt_stream(dst, out, key_id="default_key") + restored = out.getvalue() + assert restored == plain, "解密结果与明文不一致!" + print("解密还原: 一致 ✓") + + # 4. 篡改检测: 改密文一个字节, finalize 必须抛 IntegrityError + tampered = bytearray(dst.getvalue()) + tampered[len(tampered) // 2] ^= 0xFF + try: + out2 = io.BytesIO() + engine.decrypt_stream(io.BytesIO(bytes(tampered)), out2, key_id="default_key") + print("篡改检测: 未检出! (异常)") + except IntegrityError: + print("篡改检测: 捕获 IntegrityError ✓") diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/metadata.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/metadata.py new file mode 100644 index 0000000..68c14d7 --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/metadata.py @@ -0,0 +1,209 @@ +""" +元数据生成模块 (metadata) +模块:客户端 / 元数据生成 +输入:文件信息 + 加密参数 + 卷清单 +输出:init json (file_name/file_size/chunk_count/total_sha256/enc/chunks) + +init json 是服务端建任务的唯一依据 (POST /init 请求体),字段缺一不可, +组装后先自校验再输出。chunks 剥离本地 filename (服务端不关心客户端路径)。 + +用法: + cd /home/lou/文档/7z-encrypt + .venv/bin/python metadata.py [文件] [--chunk-size MB] [--key-id 密钥ID] +""" + +import argparse +import hashlib +import json +import sys +from pathlib import Path +from typing import Any + +from tqdm import tqdm + +from crypto import CryptoEngine, KeyNotFoundError +from splitter import split_stream + +SHA256_HEX_LEN = 64 +DEFAULT_CHUNK_MB = 10 + + +class MetadataError(ValueError): + """init json 字段缺失或非法""" + + +def _require_str(obj: dict[str, Any], key: str) -> str: + v = obj.get(key) + if not isinstance(v, str) or not v: + raise MetadataError(f"字段 '{key}' 缺失或非空字符串") + return v + + +def _require_pos_int(obj: dict[str, Any], key: str) -> int: + v = obj.get(key) + if not isinstance(v, int) or isinstance(v, bool) or v <= 0: + raise MetadataError(f"字段 '{key}' 缺失或非正整数") + return v + + +def _require_sha256(obj: dict[str, Any], key: str) -> str: + v = _require_str(obj, key) + if len(v) != SHA256_HEX_LEN: + raise MetadataError(f"字段 '{key}' 长度不是 {SHA256_HEX_LEN} (SHA-256 hex)") + return v + + +def build_init_json( + file_name: str, + file_size: int, + total_sha256: str, + enc: dict[str, str], + chunks: list[dict[str, Any]], +) -> dict[str, Any]: + """组装并校验 init json + + Args: + file_name: 原始文件名 + file_size: 原始文件字节数 + total_sha256: 密文整体 SHA-256 (服务端合并后验证) + enc: 加密参数 {alg, key_id, context} (来自 crypto.encrypt_stream) + chunks: 卷清单 (来自 splitter, 含 filename; 组装时剥离) + + Returns: + dict: init json {file_name, file_size, chunk_count, total_sha256, enc, chunks} + + Raises: + MetadataError: 任一必填字段缺失/非法/卷序号不连续 + """ + # enc 校验 + if not enc: + raise MetadataError("字段 'enc' 缺失或为空") + for k in ("alg", "key_id", "context"): + _require_str(enc, k) + + # chunks 校验 (序号必须 1..N 连续) + if not chunks: + raise MetadataError("字段 'chunks' 缺失或为空") + clean_chunks: list[dict[str, Any]] = [] + for i, ch in enumerate(chunks, start=1): + if ch.get("index") != i: + raise MetadataError( + f"chunks[{i}] 序号不连续 (期望 {i}, 实际 {ch.get('index')})" + ) + clean_chunks.append({ + "index": i, + "size": _require_pos_int(ch, "size"), + "sha256": _require_sha256(ch, "sha256"), + }) + + # 整体校验 + if not file_name: + raise MetadataError("file_name 缺失或为空") + if isinstance(file_size, bool) or file_size <= 0: + raise MetadataError("file_size 缺失或非法") + if len(total_sha256) != SHA256_HEX_LEN: + raise MetadataError(f"total_sha256 长度不是 {SHA256_HEX_LEN} (SHA-256 hex)") + + return { + "file_name": file_name, + "file_size": file_size, + "chunk_count": len(clean_chunks), + "total_sha256": total_sha256, + "enc": enc, + "chunks": clean_chunks, + } + + +def sha256_of(path: str | Path) -> str: + """流式计算文件 SHA-256 (不整块进内存)""" + h = hashlib.sha256() + with open(path, "rb") as f: + while True: + chunk = f.read(1 << 16) + if not chunk: + break + h.update(chunk) + return h.hexdigest() + + +def main() -> int: + ap = argparse.ArgumentParser(description="元数据生成: 加密 -> 分卷 -> 组装 init json") + ap.add_argument("file", nargs="?", default="index.mp4", help="要处理的文件 (默认 index.mp4)") + ap.add_argument("--chunk-size", type=int, default=DEFAULT_CHUNK_MB, + help=f"每卷大小 MB (默认 {DEFAULT_CHUNK_MB})") + ap.add_argument("--key-id", default="default_key", help="密钥ID (默认 default_key)") + ap.add_argument("--file-name", help="覆盖 init json 的 file_name (压缩场景用原名)") + ap.add_argument("--compressed", action="store_true", + help="标记 7z 压缩 (enc_params 加 compressed, 下载端自动解压)") + ap.add_argument("--no-encrypt-name", action="store_true", + help="不加密文件名 (默认加密成 enc: 前缀)") + args = ap.parse_args() + + src_path = Path(args.file).resolve() + if not src_path.exists(): + print(f"[错误] 文件不存在: {src_path}") + return 1 + + chunk_size = args.chunk_size * 1024 * 1024 + enc_path = src_path.with_name(src_path.name + ".enc") + manifest_path = src_path.with_name(src_path.name + "_manifest.json") + init_path = src_path.with_name(src_path.name + ".init.json") + + print("=== 元数据生成 (加密 -> 分卷 -> init json) ===") + print(f"文件: {src_path.name} ({src_path.stat().st_size / 1048576:.1f} MB), 卷大小: {args.chunk_size} MB") + + # 1. 加密 -> 加密参数 + engine = CryptoEngine() + try: + engine.get_key(args.key_id) + except KeyNotFoundError: + engine.generate_key(args.key_id) + print(f"[密钥库] 已生成新密钥 -> {engine.keyring_path}") + with open(src_path, "rb") as src, open(enc_path, "wb") as dst: + pbar = tqdm(total=src_path.stat().st_size, desc="[加密]", unit="B", + unit_scale=True, leave=False) + enc_params = engine.encrypt_stream(src, dst, args.key_id, + progress=lambda n: pbar.update(n)) + pbar.close() + print(f"[1/4] 加密完成 -> {enc_path.name} ({enc_path.stat().st_size / 1048576:.1f} MB)") + + # 2. 分卷 -> 卷清单 + with open(enc_path, "rb") as f: + pbar = tqdm(total=enc_path.stat().st_size, desc="[分卷]", unit="B", + unit_scale=True, leave=False) + manifest = split_stream(f, chunk_size, str(enc_path), + total_size=enc_path.stat().st_size, + progress=lambda n: pbar.update(n)) + pbar.close() + with open(manifest_path, "w", encoding="utf-8") as f: + json.dump(manifest, f, indent=2, ensure_ascii=False) + print(f"[2/4] 分卷完成: {len(manifest)} 卷, 清单 -> {manifest_path.name}") + + # 3. 密文整体 SHA-256 + total_sha256 = sha256_of(enc_path) + print(f"[3/4] 密文 SHA-256: {total_sha256}") + + # 4. 组装 init json + if args.compressed: + enc_params["compressed"] = True # 下载端据此 7z 解压 + file_name = args.file_name or src_path.name + if not args.no_encrypt_name: + # 元数据零知识: 文件名加密成 'enc:' 前缀 (服务端只存密文, 客户端解密还原) + file_name = engine.encrypt_name(file_name) + init_json = build_init_json( + file_name=file_name, + file_size=src_path.stat().st_size, + total_sha256=total_sha256, + enc=enc_params, + chunks=manifest, + ) + with open(init_path, "w", encoding="utf-8") as f: + json.dump(init_json, f, indent=2, ensure_ascii=False) + print(f"[4/4] init json -> {init_path.name}") + print("\n=== init json ===") + print(json.dumps(init_json, indent=2, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/splitter.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/splitter.py new file mode 100644 index 0000000..52d6892 --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/splitter.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +分卷器 (splitter) +模块:客户端 / 分卷器 +输入:密文流(文件或 stdin) +输出:N 个卷文件 + 卷清单 (index, size, sha256) + +用法: + cd /home/lou/文档/7z-encrypt + .venv/bin/python splitter.py -i index.mp4.enc -o index.mp4.enc -s 10485760 + 或: cat index.mp4.enc | .venv/bin/python splitter.py -o index.mp4.enc -s 10485760 +""" + +import argparse +import hashlib +import json +import sys +from typing import IO, Any, Callable + + +def _read_block(stream: IO[bytes], size: int) -> bytes: + """循环读取直到攒满 size 字节或 EOF。 + 管道/stdin 不保证一次 read 返回满 size, 必须循环攒块, 否则卷大小不固定。 + """ + buf = b"" + while len(buf) < size: + chunk = stream.read(size - len(buf)) + if not chunk: + break + buf += chunk + return buf + + +def split_stream( + stream: IO[bytes], + chunk_size: int, + output_prefix: str, + total_size: int | None = None, + progress: Callable[[int], Any] | None = None, +) -> list[dict[str, Any]]: + """从输入流切分卷文件, 返回清单 [{index, size, sha256, filename}] + + Args: + stream: 二进制输入流 (open(...,'rb') 或 sys.stdin.buffer) + chunk_size: 每卷最大字节数 + output_prefix: 输出前缀, 卷文件命名为 .partXXXX + total_size: 输入总大小 (进度条用, 可 None) + progress: 进度回调, 每写一卷后调 progress(已写字节) + + Returns: + list: 清单 + """ + manifest: list[dict[str, Any]] = [] + part_num = 1 + done = 0 + last_pct = -1 + label = "分卷" + + while True: + data = _read_block(stream, chunk_size) + if not data: + break + + sha256_hex = hashlib.sha256(data).hexdigest() + size = len(data) + filename = f"{output_prefix}.part{part_num:04d}" + + with open(filename, "wb") as f: + f.write(data) + + manifest.append({ + "index": part_num, + "size": size, + "sha256": sha256_hex, + "filename": filename, + }) + part_num += 1 + done += size + + # 实时进度: 有回调交给调用方 (tqdm), 否则内部打印 + if progress is not None: + progress(done) + elif total_size: + pct = done * 100 // total_size + if pct != last_pct: + last_pct = pct + sys.stdout.write( + f"\r {label}: {pct}% ({done / 1048576:.1f}/{total_size / 1048576:.1f} MB)" + ) + sys.stdout.flush() + elif part_num % 16 == 0: + sys.stdout.write(f"\r {label}: 已切 {part_num} 卷 ({done / 1048576:.1f} MB)") + sys.stdout.flush() + + if total_size: + sys.stdout.write(f"\r {label}: 100% ({total_size / 1048576:.1f} MB)\n") + sys.stdout.flush() + else: + sys.stdout.write(f"\r {label}: 完成, 共 {part_num - 1} 卷 ({done / 1048576:.1f} MB)\n") + sys.stdout.flush() + + return manifest + + +def main() -> int: + parser = argparse.ArgumentParser(description="分卷器: 把数据流切成 N 个卷, 每卷带 SHA-256") + parser.add_argument("-i", "--input", help="输入文件路径 (省略则从 stdin 读取)") + parser.add_argument("-o", "--output-prefix", required=True, + help="输出卷文件前缀 (如 'archive' -> archive.part0001)") + parser.add_argument("-s", "--chunk-size", type=int, default=10 * 1024 * 1024, + help="每卷最大字节数 (默认 10MB)") + parser.add_argument("-m", "--manifest", + help="清单 JSON 保存路径 (默认 <输出前缀>_manifest.json)") + args = parser.parse_args() + + if args.chunk_size <= 0: + print(f"[错误] chunk-size 必须大于 0, 收到: {args.chunk_size}") + return 1 + + if args.input: + with open(args.input, "rb") as f: + total = f.seek(0, 2) + f.seek(0) + manifest = split_stream(f, args.chunk_size, args.output_prefix, total_size=total) + else: + manifest = split_stream(sys.stdin.buffer, args.chunk_size, args.output_prefix) + + if not manifest: + print("[错误] 输入为空, 没有产出任何卷") + return 1 + + manifest_path = args.manifest if args.manifest else f"{args.output_prefix}_manifest.json" + with open(manifest_path, "w", encoding="utf-8") as f: + json.dump(manifest, f, indent=2, ensure_ascii=False) + + total_mb = sum(p["size"] for p in manifest) / 1048576 + print(f"分卷完成: {len(manifest)} 卷, 共 {total_mb:.1f} MB, 清单 -> {manifest_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/state.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/state.py new file mode 100644 index 0000000..2f7c25c --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/state.py @@ -0,0 +1,223 @@ +""" +任务状态模块 (state) +模块:客户端 / 任务状态 +输入:任务事件 (创建/发送成功/失败) +输出:SQLite 任务记录 + +SQLite 本地库, 重跑时查未完成任务 + 服务端 /chunks 对比出缺卷, 续传不重传。 + +表结构: + tasks task_id, file_name, file_size, chunk_count, status, created_at, updated_at + chunks task_id, idx, status, updated_at (只记录非 pending 状态, 缺行即未传) +""" + +import argparse +import sqlite3 +import sys +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +DEFAULT_DB = Path(__file__).resolve().parent / "config" / "tasks.db" + +STATUS_UPLOADING = "uploading" +STATUS_DONE = "done" +STATUS_FAILED = "failed" +CHUNK_UPLOADED = "uploaded" +CHUNK_FAILED = "failed" + + +class TaskStore: + """SQLite 任务状态存储 (每次操作独立连接, 文件锁安全)""" + + def __init__(self, db_path: str | Path = DEFAULT_DB) -> None: + self.db_path = Path(db_path) + self._init_schema() + + def _connect(self) -> sqlite3.Connection: + self.db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def _init_schema(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS tasks ( + task_id TEXT PRIMARY KEY, + file_name TEXT NOT NULL, + file_size INTEGER NOT NULL, + chunk_count INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'uploading', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS chunks ( + task_id TEXT NOT NULL, + idx INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'uploaded', + updated_at TEXT NOT NULL, + PRIMARY KEY (task_id, idx) + ); + """ + ) + + @staticmethod + def _now() -> str: + return datetime.now().isoformat(timespec="seconds") + + # ---------- 任务事件 ---------- + + def create_task( + self, task_id: str, file_name: str, file_size: int, chunk_count: int + ) -> None: + """事件: 创建任务 (状态 uploading)""" + now = self._now() + with self._connect() as conn: + conn.execute( + "INSERT INTO tasks (task_id, file_name, file_size, chunk_count, status, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (task_id, file_name, file_size, chunk_count, STATUS_UPLOADING, now, now), + ) + + def mark_chunk_uploaded(self, task_id: str, idx: int) -> None: + """事件: 卷发送成功""" + self._upsert_chunk(task_id, idx, CHUNK_UPLOADED) + + def mark_chunk_failed(self, task_id: str, idx: int) -> None: + """事件: 卷发送失败 (重试耗尽, 待补传)""" + self._upsert_chunk(task_id, idx, CHUNK_FAILED) + + def _upsert_chunk(self, task_id: str, idx: int, status: str) -> None: + now = self._now() + with self._connect() as conn: + conn.execute( + "INSERT INTO chunks (task_id, idx, status, updated_at) VALUES (?, ?, ?, ?) " + "ON CONFLICT(task_id, idx) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at", + (task_id, idx, status, now), + ) + conn.execute( + "UPDATE tasks SET updated_at = ? WHERE task_id = ?", (now, task_id) + ) + + def mark_done(self, task_id: str) -> None: + """事件: 全部完成""" + self._set_task_status(task_id, STATUS_DONE) + + def mark_failed(self, task_id: str) -> None: + """事件: 任务失败""" + self._set_task_status(task_id, STATUS_FAILED) + + def _set_task_status(self, task_id: str, status: str) -> None: + with self._connect() as conn: + conn.execute( + "UPDATE tasks SET status = ?, updated_at = ? WHERE task_id = ?", + (status, self._now(), task_id), + ) + + # ---------- 查询 ---------- + + def get_task(self, task_id: str) -> dict[str, Any] | None: + with self._connect() as conn: + row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone() + return dict(row) if row else None + + def get_chunk_status(self, task_id: str) -> dict[int, str]: + """已记录的非 pending 卷状态 {idx: uploaded/failed}""" + with self._connect() as conn: + rows = conn.execute( + "SELECT idx, status FROM chunks WHERE task_id = ?", (task_id,) + ).fetchall() + return {r["idx"]: r["status"] for r in rows} + + def get_pending_tasks(self) -> list[dict[str, Any]]: + """查询未完成任务 (重跑续传): 返回任务 + 已传卷集合""" + with self._connect() as conn: + rows = conn.execute( + "SELECT * FROM tasks WHERE status = ? ORDER BY created_at", (STATUS_UPLOADING,) + ).fetchall() + tasks = [dict(r) for r in rows] + for t in tasks: + t["uploaded_chunks"] = sorted( + idx for idx, st in self.get_chunk_status(t["task_id"]).items() + if st == CHUNK_UPLOADED + ) + return tasks + + # ---------- 清理 ---------- + + def cleanup(self, days: int = 7) -> int: + """清理超过 days 天未更新的非 uploading 任务, 返回删除数""" + cutoff = (datetime.now() - timedelta(days=days)).isoformat(timespec="seconds") + with self._connect() as conn: + rows = conn.execute( + "SELECT task_id FROM tasks WHERE status != ? AND updated_at < ?", + (STATUS_UPLOADING, cutoff), + ).fetchall() + ids = [r["task_id"] for r in rows] + for tid in ids: + conn.execute("DELETE FROM chunks WHERE task_id = ?", (tid,)) + conn.execute("DELETE FROM tasks WHERE task_id = ?", (tid,)) + return len(ids) + + +def main() -> int: + ap = argparse.ArgumentParser(description="任务状态: SQLite 本地记录 (创建/卷状态/查询/清理)") + sub = ap.add_subparsers(dest="cmd", required=True) + + p_init = sub.add_parser("init", help="创建任务") + p_init.add_argument("task_id") + p_init.add_argument("file") + p_init.add_argument("--size", type=int, required=True) + p_init.add_argument("--chunks", type=int, required=True) + + p_chunk = sub.add_parser("chunk", help="记录卷状态") + p_chunk.add_argument("task_id") + p_chunk.add_argument("index", type=int) + p_chunk.add_argument("--status", choices=["uploaded", "failed"], default="uploaded") + + p_done = sub.add_parser("done", help="标记完成") + p_done.add_argument("task_id") + + p_fail = sub.add_parser("failed", help="标记失败") + p_fail.add_argument("task_id") + + sub.add_parser("pending", help="查询未完成任务") + + p_clean = sub.add_parser("cleanup", help="清理旧任务") + p_clean.add_argument("--days", type=int, default=7) + + args = ap.parse_args() + store = TaskStore() + + if args.cmd == "init": + store.create_task(args.task_id, args.file, args.size, args.chunks) + print(f"[state] 任务 {args.task_id} 已创建 ({args.chunks} 卷)") + elif args.cmd == "chunk": + if args.status == "uploaded": + store.mark_chunk_uploaded(args.task_id, args.index) + else: + store.mark_chunk_failed(args.task_id, args.index) + print(f"[state] 卷 {args.index} -> {args.status}") + elif args.cmd == "done": + store.mark_done(args.task_id) + print(f"[state] 任务 {args.task_id} -> done") + elif args.cmd == "failed": + store.mark_failed(args.task_id) + print(f"[state] 任务 {args.task_id} -> failed") + elif args.cmd == "pending": + for t in store.get_pending_tasks(): + print( + f" {t['task_id']} {t['file_name']} " + f"{len(t['uploaded_chunks'])}/{t['chunk_count']} 卷 " + f"(创建于 {t['created_at']})" + ) + elif args.cmd == "cleanup": + n = store.cleanup(args.days) + print(f"[state] 清理 {n} 个过期任务") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py new file mode 100644 index 0000000..80710fd --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py @@ -0,0 +1,582 @@ +""" +传输客户端模块 (transfer) +模块:客户端 / 传输客户端 +输入:init json + 卷文件 + 服务器配置 +输出:服务端回执 (status/file_id) + +协议 (共享契约): + POST /api/transfer/init {init json} -> {transfer_id} + GET /api/transfer/{id}/chunks -> {received: [n...]} + PUT /api/transfer/{id}/chunk/{n} 卷密文 -> {ok} + POST /api/transfer/{id}/complete -> {status, file_id} + +断点续传: 先查已收卷集合只补缺卷; 单卷失败重试 max_retries 次, 仍失败抛 TransferError。 +""" + +import argparse +import base64 +import json +import os +import shutil +import subprocess +import sys +import time +from pathlib import Path +from typing import IO, Any + +from tqdm import tqdm +from urllib import error as urlerror +from urllib import parse as urlparse +from urllib import request as urlrequest + +DEFAULT_MAX_RETRIES = 3 +DEFAULT_RETRY_DELAY = 1.0 # 秒 + + +class TransferError(RuntimeError): + """传输失败 (服务端错误/网络错误/重试耗尽)""" + + +class TransferClient: + """HTTP 分片上传客户端 (标准库 urllib, 零依赖)""" + + def __init__( + self, + base_url: str, + timeout: int | float = 30, + max_retries: int = DEFAULT_MAX_RETRIES, + retry_delay: float = DEFAULT_RETRY_DELAY, + token: str = "", + ) -> None: + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self.max_retries = max_retries + self.retry_delay = retry_delay + self.token = token + + # ---------- HTTP 封装 ---------- + + def _request( + self, method: str, path: str, body: bytes | None = None, timeout: int | float | None = None + ) -> tuple[int, dict[str, Any]]: + """发请求, 返回 (HTTP状态码, JSON 载荷)。4xx/5xx 不抛, 由调用方判断""" + req = urlrequest.Request(self.base_url + path, data=body, method=method) + if body is not None: + req.add_header("Content-Type", "application/octet-stream") + if self.token: + req.add_header("Authorization", f"Bearer {self.token}") + try: + with urlrequest.urlopen(req, timeout=timeout or self.timeout) as resp: + data = resp.read() + code = resp.status + except urlerror.HTTPError as e: + code = e.code + data = e.read() + except urlerror.URLError as e: + raise TransferError(f"网络错误: {e.reason}") from e + except TimeoutError: + # urllib 读响应阶段的超时是裸抛, 不包 URLError + raise TransferError(f"请求超时 ({timeout or self.timeout}s)") from None + try: + payload: dict[str, Any] = json.loads(data) if data else {} + except json.JSONDecodeError: + payload = {} + return code, payload + + # ---------- 协议端点 ---------- + + def whoami(self) -> str: + """GET /api/auth/me: 验证 token 并返回当前用户名""" + code, payload = self._request("GET", "/api/auth/me") + if code != 200: + raise TransferError(f"登录验证失败 (HTTP {code}): {payload}") + return str(payload.get("username", "?")) + + def register(self, username: str, password: str) -> dict[str, Any]: + """POST /api/auth/register: 注册账号 (成功即登录)""" + code, payload = self._request( + "POST", "/api/auth/register", + json.dumps({"username": username, "password": password}).encode(), + ) + if code != 200 or "token" not in payload: + raise TransferError(f"注册失败 (HTTP {code}): {payload}") + return payload + + def login(self, username: str, password: str) -> dict[str, Any]: + """POST /api/auth/login: 登录, 返回 {username, token}""" + code, payload = self._request( + "POST", "/api/auth/login", + json.dumps({"username": username, "password": password}).encode(), + ) + if code != 200 or "token" not in payload: + raise TransferError(f"登录失败 (HTTP {code}): {payload}") + return payload + + def logout(self) -> None: + """POST /api/auth/logout: 注销当前 token""" + code, payload = self._request("POST", "/api/auth/logout") + if code != 200: + raise TransferError(f"注销失败 (HTTP {code}): {payload}") + + def init_transfer(self, init_json: dict[str, Any]) -> str: + """POST /init: 建任务, 返回 transfer_id""" + code, payload = self._request( + "POST", "/api/transfer/init", json.dumps(init_json).encode() + ) + if code != 200 or "transfer_id" not in payload: + raise TransferError(f"init 失败 (HTTP {code}): {payload}") + return payload["transfer_id"] + + def get_received_chunks(self, transfer_id: str) -> set[int]: + """GET /chunks: 查询服务端已收卷 (断点续传依据)""" + code, payload = self._request("GET", f"/api/transfer/{transfer_id}/chunks") + if code != 200: + raise TransferError(f"查询已收卷失败 (HTTP {code})") + return set(payload.get("received", [])) + + def upload_chunk(self, transfer_id: str, index: int, chunk_path: Path) -> None: + """PUT /chunk/{n}: 上传单卷, 失败重试 max_retries 次""" + data = chunk_path.read_bytes() + code: int = 0 + for attempt in range(1, self.max_retries + 1): + code, payload = self._request( + "PUT", f"/api/transfer/{transfer_id}/chunk/{index}", body=data + ) + if code == 200 and payload.get("ok"): + return + if attempt < self.max_retries: + time.sleep(self.retry_delay) + raise TransferError( + f"卷 {index} 上传失败 (HTTP {code}, 重试 {self.max_retries} 次耗尽)" + ) + + def complete(self, transfer_id: str) -> dict[str, Any]: + """POST /complete: 触发服务端汇聚合并入库, 返回回执 + + 服务端零合并 (卷齐即入库), 正常秒回; 超时 600s 兜底。 + """ + sys.stdout.write("[等待] 服务端确认入库...\n") + sys.stdout.flush() + code, payload = self._request( + "POST", f"/api/transfer/{transfer_id}/complete", timeout=600 + ) + if code != 200: + raise TransferError(f"complete 失败 (HTTP {code}): {payload}") + return payload + + def list_files(self) -> list[dict[str, Any]]: + """GET /api/files: 服务端文件列表 (ls)""" + code, payload = self._request("GET", "/api/files") + if code != 200: + raise TransferError(f"文件列表失败 (HTTP {code}): {payload}") + return payload.get("files", []) + + def delete_file(self, file_id: str) -> None: + """DELETE /api/files/{id}: 删除服务端文件""" + code, payload = self._request("DELETE", f"/api/files/{file_id}") + if code != 200: + raise TransferError(f"删除失败 (HTTP {code}): {payload}") + if not payload.get("ok"): + raise TransferError(f"删除失败: {payload}") + + def get_quota(self) -> dict[str, Any]: + """GET /api/quota: 空间配额 (used/quota/remain/percent)""" + code, payload = self._request("GET", "/api/quota") + if code != 200: + raise TransferError(f"配额查询失败 (HTTP {code}): {payload}") + return payload + + def download(self, file_id: str, dest: Path | None = None) -> tuple[Path, dict[str, Any]]: + """逐卷下载密文 (零合并: 服务端只存卷, 客户端本地拼接), 返回 (密文路径, 解密参数)。 + + 先拉卷 1 拿响应头 (X-Enc-Params/X-Chunk-Count/文件名), 再循环 2..N 追加写。 + dest 缺省用服务端文件名; 是目录则拼文件名。 + """ + try: + with self._open_chunk(file_id, 1) as resp: + enc_params: dict[str, Any] = {} + ep = resp.headers.get("X-Enc-Params") or "" + if ep: + enc_params = json.loads(base64.b64decode(ep).decode()) + chunk_count = max(int(resp.headers.get("X-Chunk-Count") or 1), 1) + name = _parse_disposition(resp.headers.get("Content-Disposition") or "") + enc_params.setdefault("file_name", name) + if dest is None: + dest = Path(name) + elif dest.is_dir(): + dest = dest / name + assert dest is not None + outer = tqdm(total=chunk_count, desc="[下载]", unit="卷", + position=0, leave=False) + with open(dest, "wb") as f: + self._download_chunk_to(resp, f, outer, 1, chunk_count) + for idx in range(2, chunk_count + 1): + with self._open_chunk(file_id, idx) as resp: + with open(dest, "ab") as f: + self._download_chunk_to(resp, f, outer, idx, chunk_count) + outer.close() + return dest, enc_params + except urlerror.HTTPError as e: + if e.code == 401: + raise TransferError("未授权: token 无效 (检查 config server.token)") from e + if e.code == 404: + raise TransferError("文件或卷不存在 (HTTP 404)") from e + raise TransferError(f"下载失败 (HTTP {e.code})") from e + except urlerror.URLError as e: + raise TransferError(f"网络错误: {e.reason}") from e + except TimeoutError: + raise TransferError(f"下载超时 ({self.timeout}s)") from None + + def _open_chunk(self, file_id: str, idx: int): + """打开单卷响应流 (带认证)""" + req = urlrequest.Request( + self.base_url + f"/api/files/{file_id}/chunk/{idx}", method="GET" + ) + if self.token: + req.add_header("Authorization", f"Bearer {self.token}") + return urlrequest.urlopen(req, timeout=self.timeout) + + def _show_chunk_progress(self, cur: int, total_chunks: int) -> None: + """按卷粒度显示下载进度 (与上传日志对称)""" + pct = cur * 100 // total_chunks + sys.stdout.write(f"\r[下载] 卷 {cur}/{total_chunks} ({pct}%)") + sys.stdout.flush() + + def _download_chunk_to( + self, resp, f: IO[bytes], outer: Any, idx: int, total_chunks: int + ) -> int: + """卷内字节流写入 + 进度 (内层字节条), 返回写入字节数""" + size = int(resp.headers.get("Content-Length") or 0) + inner = tqdm(total=size, desc=f" 卷 {idx}/{total_chunks}", unit="B", + unit_scale=True, position=1, leave=False) + done = 0 + while True: + chunk = resp.read(1 << 20) + if not chunk: + break + f.write(chunk) + done += len(chunk) + inner.update(len(chunk)) + inner.close() + outer.update(1) + return done + + # ---------- 全流程 ---------- + + def transfer( + self, init_json: dict[str, Any], chunk_files: dict[int, Path] + ) -> dict[str, Any]: + """init -> 查已收 -> 逐卷上传(跳过已收) -> complete + + Args: + init_json: metadata.build_init_json 的产物 + chunk_files: {卷序号: 卷文件路径} + + Returns: + dict: 服务端回执 {status, file_id} + """ + transfer_id = self.init_transfer(init_json) + print(f"[传输] transfer_id: {transfer_id}") + + received = self.get_received_chunks(transfer_id) + total = init_json["chunk_count"] + pending = [i for i in range(1, total + 1) if i not in received] + if received: + print(f"[传输] 断点续传: 服务端已收 {len(received)}/{total} 卷, 补传 {len(pending)} 卷") + + pbar = tqdm(total=len(pending), desc="[传输] 上传", unit="卷", leave=False) + for done, index in enumerate(pending, start=1): + try: + self.upload_chunk(transfer_id, index, chunk_files[index]) + except TransferError: + raise # 重试耗尽, 由调用方决定 (记录断点状态待补传) + pbar.update(1) + pbar.set_description(f"[传输] 上传 {done}/{len(pending)} 卷") + pbar.close() + + receipt = self.complete(transfer_id) + print(f"[传输] 完成: {receipt}") + return receipt + + +def _safe_name(name: str) -> str: + """下载文件名消毒: 只保留 basename, 去路径穿越/控制字符 (防恶意服务端)""" + name = Path(name).name # 去掉目录部分 (.. / ../../ 无效化) + return "".join(c for c in name if c.isprintable() and c not in "/\\\x00").strip() or "download.bin" + + +def _decrypt_name(token: str) -> str | None: + """解密 'enc:' 前缀文件名, 失败返回 None (明文名原样返回)""" + if not token.startswith("enc:"): + return None + try: + from crypto import CryptoEngine + return CryptoEngine(_load_keyring_path()).decrypt_name(token) + except Exception: + return None + + +def _parse_disposition(cd: str) -> str: + """解析 Content-Disposition 文件名 (utf-8 优先, 解密 enc: 前缀)""" + name = "download.bin" + if "filename*=utf-8''" in cd: + name = urlparse.unquote(cd.split("filename*=utf-8''")[1].split(";")[0]) + elif "filename=" in cd: + name = cd.split("filename=")[1].split(";")[0].strip('"') + plain = _decrypt_name(name) + return _safe_name(plain if plain is not None else name) + + +def _config_path() -> Path: + """配置文件路径: SZ_CONFIG env -> XDG (~/.config/7z-encrypt/config.json) + -> 兼容旧式相对 config/config.json (源码开发模式) + + 安装版 (deb) cwd 不可写, 配置必须走用户级 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("config/config.json") + return xdg_cfg if not legacy.exists() else legacy + + +def _config_server() -> str: + """从配置文件读 server.url (安装版 CLI 默认服务器)""" + try: + cfg = _config_path() + return str(json.loads(cfg.read_text(encoding="utf-8")).get("server", {}).get("url", "")) + except (OSError, json.JSONDecodeError): + return "" + + +def _load_token() -> str: + """从配置文件读 server.token (CLI 默认来源, XDG 优先)""" + try: + cfg = _config_path() + return str(json.loads(cfg.read_text(encoding="utf-8")).get("server", {}).get("token", "")) + except (OSError, json.JSONDecodeError): + return "" + + +def _save_token(token: str) -> None: + """登录/注册成功后 token 持久化到配置文件 server.token""" + cfg = _config_path() + cfg.parent.mkdir(parents=True, exist_ok=True) + data = json.loads(cfg.read_text(encoding="utf-8")) if cfg.exists() else {} + data.setdefault("server", {})["token"] = token + cfg.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + + +def _load_keyring_path() -> Path: + """密钥库路径: 配置显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)""" + try: + cfg = _config_path() + raw = str(json.loads(cfg.read_text(encoding="utf-8")).get("crypto", {}).get("keyring_path", "")) + if raw: + return Path(raw) + except (OSError, json.JSONDecodeError): + pass + from crypto import DEFAULT_KEYRING + return DEFAULT_KEYRING + + +def main() -> int: + ap = argparse.ArgumentParser(description="传输客户端: 上传 / 文件列表 / 下载") + ap.add_argument("--server", default=None, help="服务端地址 (默认读配置 server.url)") + ap.add_argument("--init", help="init json 路径 (上传模式)") + ap.add_argument("--manifest", help="卷清单路径 (上传模式)") + ap.add_argument("--list", action="store_true", help="列出服务端文件") + ap.add_argument("--quota", action="store_true", help="显示空间配额") + ap.add_argument("--download", metavar="FILE_ID", help="下载文件") + ap.add_argument("--delete", metavar="FILE_ID", help="删除服务端文件") + ap.add_argument("--out", help="保存路径 (目录或完整文件名)") + ap.add_argument("--decrypt", action="store_true", help="下载后用本地 keyring 解密还原明文") + ap.add_argument("--token", help="服务端 token (默认读 config/config.json)") + ap.add_argument("--register", nargs=2, metavar=("用户名", "密码"), help="注册账号 (成功即登录)") + ap.add_argument("--login", nargs=2, metavar=("用户名", "密码"), help="登录账号") + ap.add_argument("--logout", action="store_true", help="注销当前账号 (token 失效)") + ap.add_argument("--whoami", action="store_true", help="显示当前登录用户") + args = ap.parse_args() + + server = args.server or _config_server() + if not server: + ap.error("未指定 --server 且配置无 server.url (先 sz-config --set-server)") + token = args.token if args.token is not None else _load_token() + client = TransferClient(server, token=token) + + if args.whoami: + try: + uname = client.whoami() + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[用户] {uname}") + return 0 + + if args.register: + try: + payload = client.register(args.register[0], args.register[1]) + _save_token(payload["token"]) + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[注册] 成功, 已登录: {payload['username']} (token 已保存)") + return 0 + + if args.login: + try: + payload = client.login(args.login[0], args.login[1]) + _save_token(payload["token"]) + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[登录] 成功: {payload['username']} (token 已保存)") + return 0 + + if args.logout: + try: + client.logout() + _save_token("") + except TransferError as e: + print(f"[错误] {e}") + return 1 + print("[注销] 已退出 (token 已清除)") + return 0 + + if args.list: + try: + files = client.list_files() + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[文件] 共 {len(files)} 个:") + for f in files: + size_mb = f["size"] / 1048576 + name = _decrypt_name(f["file_name"]) or f["file_name"] + print(f" {f['file_id']} {name} ({size_mb:.1f} MB) {f['created_at']}") + return 0 + + if args.download: + try: + dest_arg = Path(args.out) if args.out else None + if args.decrypt: + # 密文先落临时 .enc, 本地解密后删 + if dest_arg and not dest_arg.is_dir(): + tmp_enc = Path(str(dest_arg) + ".enc") + else: + tmp_enc = (dest_arg if dest_arg else Path(".")) / ".sz-download.enc" + enc_path, enc_params = client.download(args.download, tmp_enc) + if enc_params.get("key_id") is None: + print("[错误] 响应缺解密参数 (X-Enc-Params)") + return 1 + name = str(enc_params.get("file_name", "download.bin")) + if dest_arg and dest_arg.is_dir(): + final = dest_arg / name + elif dest_arg: + final = dest_arg + else: + final = Path(name) + from crypto import CryptoEngine + engine = CryptoEngine(_load_keyring_path()) + if enc_params.get("compressed"): + # 解密还原的是 7z 流 -> 解压成原始文件 + tmp7z = tmp_enc.with_suffix(".7z") + pbar = tqdm(total=enc_path.stat().st_size, desc="[解密]", unit="B", + unit_scale=True, leave=False) + with open(enc_path, "rb") as src, open(tmp7z, "wb") as dst: + engine.decrypt_stream( + src, dst, + enc_params["key_id"], + enc_params.get("context", "7z-encrypt:v1").encode(), + progress=lambda n: pbar.update(n), + ) + pbar.close() + enc_path.unlink() + if dest_arg and not dest_arg.is_dir(): + out_dir = dest_arg.parent + else: + out_dir = (dest_arg if dest_arg else Path(".")) + out_dir.mkdir(parents=True, exist_ok=True) + print(f"[解压] 7z x -o{out_dir} ...") + r = subprocess.run( + ["7z", "x", "-y", "-bd", f"-o{out_dir}", str(tmp7z)], + capture_output=True, text=True, + ) + if r.returncode != 0: + print(f"[错误] 7z 解压失败: {r.stderr[-300:]}") + return 1 + tmp7z.unlink() + if dest_arg and not dest_arg.is_dir(): + # 指定了输出文件名: 包内原名 -> 改名为目标 + shutil.move(out_dir / name, dest_arg) + final = dest_arg + print(f"[下载] 完成 -> {final} ({final.stat().st_size / 1048576:.1f} MB, 已解密+解压)") + else: + pbar = tqdm(total=enc_path.stat().st_size, desc="[解密]", unit="B", + unit_scale=True, leave=False) + with open(enc_path, "rb") as src, open(final, "wb") as dst: + engine.decrypt_stream( + src, dst, + enc_params["key_id"], + enc_params.get("context", "7z-encrypt:v1").encode(), + progress=lambda n: pbar.update(n), + ) + pbar.close() + enc_path.unlink() + print(f"[下载] 完成 -> {final} ({final.stat().st_size / 1048576:.1f} MB, 已解密)") + else: + dest, _ = client.download(args.download, dest_arg) + print(f"[下载] 完成 -> {dest} ({dest.stat().st_size / 1048576:.1f} MB, 密文)") + except TransferError as e: + print(f"[错误] {e}") + return 1 + return 0 + + if args.quota: + try: + q = client.get_quota() + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[配额] 已用 {q['used_bytes'] / 1048576:.1f} MB / " + f"{q['quota_bytes'] / 1048576:.1f} MB ({q['percent']}%) " + f"剩余 {q['remain_bytes'] / 1048576:.1f} MB") + return 0 + + if args.delete: + try: + client.delete_file(args.delete) + except TransferError as e: + print(f"[错误] {e}") + return 1 + print(f"[删除] 已删除: {args.delete}") + return 0 + + if not (args.init and args.manifest): + ap.error("需指定 --list / --quota / --download / --delete / (--init + --manifest) 之一") + + init_json = json.loads(Path(args.init).read_text(encoding="utf-8")) + manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8")) + chunk_files = { + ch["index"]: Path(ch["filename"]).resolve() + for ch in manifest + } + missing = [i for i, p in chunk_files.items() if not p.exists()] + if missing: + print(f"[错误] 卷文件缺失: {missing}") + return 1 + + try: + client.transfer(init_json, chunk_files) + except TransferError as e: + print(f"[错误] {e}") + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/tui.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/tui.py new file mode 100644 index 0000000..9f10516 --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/tui.py @@ -0,0 +1,510 @@ +#!/usr/bin/env python3 +"""7z-encrypt 客户端 TUI (Python input() REPL, 中文 IME 友好) + +vue-tui 版输入竞争问题修复前的过渡方案 (mt-translate 同款模式: +vue-tui 的 raw mode 输入无法配合 IME, Python cooked-mode input() 可以)。 + +流程: 直接驱动 7z-encrypt 客户端模块 CLI, 实时透传进度输出。 +""" + +import glob +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from urllib import request as urlrequest + +CLIENT_DIR = os.environ.get("SZ_CLIENT_DIR", "/home/lou/文档/7z-encrypt") +PYTHON = f"{CLIENT_DIR}/.venv/bin/python" +# 编译产物优先 (Nuitka 原生二进制), 无则回退源码解释执行 +TRANSFER_BIN = f"{CLIENT_DIR}/bin/sz-transfer" +TRANSFER_CMD = [TRANSFER_BIN] if os.path.exists(TRANSFER_BIN) else [PYTHON, "transfer.py"] + + +def _tui_config() -> Path: + """配置文件: 源码目录 config/config.json (开发) -> XDG (安装版)""" + p = Path(f"{CLIENT_DIR}/config/config.json") + if p.exists(): + return p + xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config") + return Path(xdg) / "7z-encrypt" / "config.json" + + +def _default_server() -> str: + """从配置文件读 server.url (无则本地默认)""" + try: + cfg = json.loads(_tui_config().read_text(encoding="utf-8")) + return str(cfg.get("server", {}).get("url", "http://127.0.0.1:8000")) + except (OSError, json.JSONDecodeError): + return "http://127.0.0.1:8000" + + +def run(args: list[str]) -> int: + """子进程运行 7z-encrypt CLI, 实时透传输出 (进度可见) + + transfer.py 走编译产物 (sz-transfer 原生二进制), 其余模块走 venv python + """ + if args and args[0] == "transfer.py": + cmd = TRANSFER_CMD + args[1:] + else: + cmd = [PYTHON] + args + proc = subprocess.Popen( + cmd, + cwd=CLIENT_DIR, + text=True, + encoding="utf-8", + ) + try: + return proc.wait() + except KeyboardInterrupt: + proc.kill() + print("\n[已中断]") + return 130 + + +def _clean(s: str) -> str: + """剥掉包裹引号 (用户复制路径常带 ' 或 ")""" + return s.strip().strip("'\"") if s else "" + + +def cleanup_artifacts(path: str) -> None: + """上传成功后清理临时产物 (密文/分卷/init json/清单) + + 失败时不清: 断点续传依赖卷文件 + init json。 + """ + removed = 0 + for pattern in ( + path + ".enc", + path + ".enc.part*", + path + ".init.json", + path + "_manifest.json", + ): + for p in glob.glob(pattern): + try: + os.remove(p) + removed += 1 + except OSError: + pass + if removed: + print(f"[清理] 已删除 {removed} 个临时文件") + + +def upload() -> None: + print("--- 上传文件 ---") + path = _clean(input("文件路径: ")) + if not path or not os.path.exists(path): + print(f"[错误] 文件不存在: {path}") + return + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server # 只输 IP:端口 自动补协议 + chunk_size = _clean(input("卷大小 MB [10]: ")) or "10" + + print("\n[1/3] 7z 压缩 (-mx=9) ...") + src7z = path + ".7z" + r = subprocess.run( + ["7z", "a", "-mx=9", "-bd", src7z, path], + capture_output=True, text=True, + ) + if r.returncode != 0: + print(f"[错误] 7z 压缩失败: {r.stderr[-300:]}") + return + print(f"[1/3] 压缩完成: {os.path.getsize(path) / 1048576:.1f} MB -> {os.path.getsize(src7z) / 1048576:.1f} MB") + + print("\n[2/3] 加密 + 分卷 + 元数据 ...") + code = run([ + "metadata.py", src7z, "--chunk-size", chunk_size, + "--file-name", os.path.basename(path), "--compressed", + ]) + if code != 0: + os.remove(src7z) + print("[失败] 加密/分卷失败") + return + + print("\n[3/3] 上传 ...") + init_json = src7z + ".init.json" + manifest = src7z + "_manifest.json" + code = run(["transfer.py", "--server", server, "--init", init_json, "--manifest", manifest]) + if code == 0: + print("[完成]") + cleanup_artifacts(src7z) # 清 .enc/.part/init/manifest + os.remove(src7z) # 清压缩临时文件 + else: + print("[失败] 上传出错 (加密产物保留, 可续传)") + os.remove(src7z) # 压缩临时文件无续传价值, 清掉 + + +def tasks() -> None: + print("--- 任务续传 ---") + run(["state.py", "pending"]) + + +def config_show() -> None: + print("--- 配置 ---") + run(["config.py"]) + + +def config_edit() -> None: + url = _clean(input("服务器地址: ")) + if url: + run(["config.py", "--set-server", url]) + token = _clean(input("认证 token (留空不改): ")) + if token: + run(["config.py", "--set-token", token]) + config_show() + + +def auth_register() -> None: + """注册账号 (成功即登录)""" + print("--- 注册账号 ---") + user = _clean(input("用户名 (3-32 位字母数字或下划线): ")) + pw = input("密码 (至少 6 位): ") + if not user or not pw: + print("[错误] 用户名/密码不能为空") + return + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--register", user, pw]) + print("[完成]" if code == 0 else "[失败] 注册出错") + + +def auth_login() -> None: + """登录账号""" + print("--- 登录 ---") + user = _clean(input("用户名: ")) + pw = input("密码: ") + if not user or not pw: + print("[错误] 用户名/密码不能为空") + return + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--login", user, pw]) + print("[完成]" if code == 0 else "[失败] 登录出错") + + +def auth_logout() -> None: + """注销当前账号 (token 失效)""" + print("--- 注销 ---") + c = input("确认退出当前账号? (输入 yes): ").strip().lower() + if c != "yes": + print("[取消]") + return + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--logout"]) + print("[完成] 已注销" if code == 0 else "[失败] 注销出错") + + +def regenerate_key() -> None: + """重新生成密钥: 红色风险警告 + 二次确认 (旧密文全废)""" + print("--- 重新生成密钥 ---") + 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 · 建议先导出旧密钥备份 (菜单 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 + c2 = input("再次确认: 输入 CONFIRM 才执行: ").strip() + if c2 != "CONFIRM": + print("[取消] 未重新生成") + return + code = run(["config.py", "--regenerate-key", "--yes"]) + print("[完成]" if code == 0 else "[失败] 重新生成出错") + + +def files_list() -> None: + """ls: 列出服务端文件 + 空间配额""" + print("--- 文件列表 ---") + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--list"]) + if code == 0: + run(["transfer.py", "--server", server, "--quota"]) + if code != 0: + print("[失败] 获取列表出错") + + +def download() -> None: + """下载: 先列文件, 再选 file_id, 本地解密还原""" + print("--- 下载文件 ---") + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--list"]) + if code != 0: + print("[失败] 获取列表出错, 无法下载") + return + file_id = _clean(input("file_id: ")) + if not file_id: + return + out = _clean(input("保存路径 [当前目录/服务端文件名]: ")) + args = ["transfer.py", "--server", server, "--download", file_id, "--decrypt"] + if out: + args += ["--out", out] + code = run(args) + print("[完成]" if code == 0 else "[失败] 下载出错") + + +def delete_file() -> None: + """删除: 先列文件, 再选 file_id 删除""" + print("--- 删除文件 ---") + server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() + if not server.startswith(("http://", "https://")): + server = "http://" + server + code = run(["transfer.py", "--server", server, "--list"]) + if code != 0: + print("[失败] 获取列表出错, 无法删除") + return + file_id = _clean(input("要删除的 file_id: ")) + if not file_id: + return + code = run(["transfer.py", "--server", server, "--delete", file_id]) + print("[完成]" if code == 0 else "[失败] 删除出错") + + +def _keyring_path() -> Path: + """密钥库路径: 配置显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)""" + try: + cfg = json.loads(_tui_config().read_text(encoding="utf-8")) + raw = cfg.get("crypto", {}).get("keyring_path", "") + if raw: + return Path(raw) + except (OSError, json.JSONDecodeError): + pass + env = os.environ.get("SZ_KEYRING") + if env: + return Path(env) + xdg = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share") + return Path(xdg) / "7z-encrypt" / "keyring.json" + + +def export_key() -> None: + """密钥导出: 红色风险警告 + 二次确认 (防误操作)""" + print("--- 密钥导出 ---") + 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 · 任何人拿到此文件都能解密你服务器上的全部密文\033[0m") + print("\033[91m · 密钥泄露 = 数据完全暴露, 无法撤销\033[0m") + print("\033[91m · 请确保导出文件安全存放, 不要发到聊天/网盘/邮件\033[0m") + print("\033[91m · 用途: 备份 / 迁移到新设备 (新设备导入后可解密)\033[0m") + print("\033[91m" + "═" * 52 + "\033[0m") + print() + c1 = input("是否了解风险并继续? (输入 yes 继续): ").strip().lower() + if c1 != "yes": + print("[取消] 未导出") + return + c2 = input("再次确认: 输入 CONFIRM 才执行导出: ").strip() + if c2 != "CONFIRM": + print("[取消] 未导出") + return + keyring = _keyring_path() + if not keyring.exists(): + print(f"[错误] 密钥文件不存在: {keyring}") + return + out = _clean(input(f"导出到路径 [~/keyring-export.json]: ")) + out = out or str(Path.home() / "keyring-export.json") + shutil.copy2(keyring, out) + os.chmod(out, 0o600) # 密钥文件权限 600 + print(f"[完成] 密钥已导出 -> {out} (权限 600)") + print("[提示] 新设备导入: 将该文件覆盖到客户端 config/keyring.json 即可") + + +def import_key() -> None: + """密钥导入: 红色风险警告 + 二次确认 + 自动备份现有密钥库""" + print("--- 密钥导入 ---") + src = _clean(input("密钥文件路径: ")) + if not src or not os.path.exists(src): + print("[错误] 文件不存在") + return + try: + with open(src, encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict) or not data: + print("[错误] 不是有效的密钥库文件 (期望 {key_id: hex})") + return + for kid, val in data.items(): + if not isinstance(val, str) or len(val) != 64: + print(f"[错误] 密钥 {kid} 格式非法 (期望 32 字节 hex)") + return + except (OSError, ValueError) as e: + print(f"[错误] 无法读取密钥文件: {e}") + return + + keyring = _keyring_path() + 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 · 除非你确认此文件就是你要用的密钥\033[0m") + print("\033[91m · 导入前会自动备份当前密钥 -> keyring.json.bak\033[0m") + print("\033[91m · 备份文件请勿删除, 出错可手动恢复\033[0m") + print("\033[91m" + "═" * 52 + "\033[0m") + print() + c1 = input("是否了解风险并继续? (输入 yes 继续): ").strip().lower() + if c1 != "yes": + print("[取消] 未导入") + return + c2 = input("再次确认: 输入 CONFIRM 才执行导入: ").strip() + if c2 != "CONFIRM": + print("[取消] 未导入") + return + + # 备份现有密钥库 -> keyring.json.bak (防手滑可回滚) + if keyring.exists(): + shutil.copy2(keyring, str(keyring) + ".bak") + print(f"[备份] 现有密钥已备份 -> config/keyring.json.bak") + shutil.copy2(src, keyring) + os.chmod(keyring, 0o600) + print(f"[完成] 密钥已导入 (来自 {src}, 权限 600)") + print("[提示] 导入后即可解密该密钥加密的密文; 出错可恢复 keyring.json.bak") + + +def _load_token() -> str: + """从配置文件读 server.token (源码目录优先, 安装版走 XDG)""" + try: + cfg = json.loads(_tui_config().read_text(encoding="utf-8")) + return str(cfg.get("server", {}).get("token", "")) + except (OSError, json.JSONDecodeError): + return "" + + +def _check_login() -> str | None: + """验证当前 token: 有效返回用户名, 否则 None (不登录不能操作文件)""" + server = _default_server() + token = _load_token() + if not server or not token: + return None + try: + req = urlrequest.Request( + server.rstrip("/") + "/api/auth/me", + headers={"Authorization": f"Bearer {token}"}, + ) + with urlrequest.urlopen(req, timeout=10) as resp: + return str(json.loads(resp.read()).get("username", "")) + except Exception: + return None + + +def _require_login() -> bool: + """文件操作前置校验: 未登录直接报错 (不登入不能进行文件操作)""" + if _check_login() is not None: + return True + print("\033[91m[错误] 未登录, 不能进行文件操作, 请先登录 (菜单 10)\033[0m") + return False + + +def main() -> int: + print("7z-encrypt 客户端 (Python TUI)") + print("安全文件传输 · 加密 / 分卷 / 上传") + while True: + user = _check_login() + print() + if user is None: + # 未登录: 文件操作全部不可用 (灰掉), 只有登录/注册/配置 + print("=== 请先登录 ===") + print(f"服务器: {_default_server() or '(未配置)'}") + print(" 1. 登录") + print(" 2. 注册") + print(" 3. 配置服务器") + print(" q. 退出") + c = input("> ").strip().lower() + if c == "1": + auth_login() + elif c == "2": + auth_register() + elif c == "3": + config_edit() + elif c == "q": + print("再见") + return 0 + else: + print("\033[91m[错误] 未登录, 只能进行 登录/注册/配置 操作\033[0m") + continue + + # 已登录: 完整主菜单 + print(f"主菜单 (当前用户: {user})") + print(" 1. 上传文件") + print(" 2. 任务续传") + print(" 3. 配置") + print(" 4. 文件列表 (ls)") + print(" 5. 下载文件") + print(" 6. 删除文件") + print(" 7. 密钥导出") + print(" 8. 密钥导入") + print(" 9. 注册账号") + print(" 10. 登录") + print(" 11. 注销") + print(" 12. 重新生成密钥") + print(" q. 退出") + choice = input("> ").strip().lower() + if choice == "1": + if not _require_login(): + continue + upload() + elif choice == "2": + if not _require_login(): + continue + tasks() + elif choice == "3": + config_show() + edit = input("修改服务器地址? (y/N): ").strip().lower() + if edit == "y": + config_edit() + elif choice == "4": + if not _require_login(): + continue + files_list() + elif choice == "5": + if not _require_login(): + continue + download() + elif choice == "6": + if not _require_login(): + continue + delete_file() + elif choice == "7": + export_key() + elif choice == "8": + import_key() + elif choice == "9": + auth_register() + elif choice == "10": + auth_login() + elif choice == "11": + auth_logout() + continue # 注销后回未登录界面 + elif choice == "12": + regenerate_key() + elif choice == "q": + print("再见") + return 0 + else: + print("[提示] 请选择 1-12 或 q") + + +if __name__ == "__main__": + try: + sys.exit(main()) + except (KeyboardInterrupt, EOFError): + print("\n再见") + sys.exit(0) diff --git a/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/changelog.gz b/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/changelog.gz new file mode 100644 index 0000000..df13599 Binary files /dev/null and b/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/changelog.gz differ diff --git a/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/copyright b/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/copyright new file mode 100644 index 0000000..7d292b7 --- /dev/null +++ b/debian/7z-encrypt-client/usr/share/doc/7z-encrypt-client/copyright @@ -0,0 +1,27 @@ +Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/ +Upstream-Name: 7z-encrypt +Upstream-Contact: edgevoid +Source: https://gitee.com/edgevoid/7z-encrypt + +Files: * +Copyright: 2026 edgevoid +License: MIT + +License: MIT + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + . + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + . + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE. diff --git a/debian/changelog b/debian/changelog new file mode 100644 index 0000000..8cad0fd --- /dev/null +++ b/debian/changelog @@ -0,0 +1,7 @@ +7z-encrypt-client (1.0.0) unstable; urgency=medium + + * 首个 Termux/PRoot 适配版: 配置 XDG 化 + venv 依赖自建 + * sz-transfer / sz-tui / sz-config 三个命令 + * 零知识 + 零合并 + 文件名加密 + 账号体系 + 登录门禁 + + -- edgevoid Mon, 10 Aug 2026 16:00:00 +0800 diff --git a/debian/control b/debian/control new file mode 100644 index 0000000..5aa69da --- /dev/null +++ b/debian/control @@ -0,0 +1,21 @@ +Source: 7z-encrypt-client +Section: utils +Priority: optional +Maintainer: edgevoid +Build-Depends: debhelper-compat (= 13) +Standards-Version: 4.6.2 + +Package: 7z-encrypt-client +Architecture: all +Depends: python3 (>= 3.10), python3-venv +Description: 端到端加密文件传输客户端 (Termux/PRoot 适配版) + 零知识 + 零合并的文件安全传输客户端: + 7z 压缩 -> cobblestone 加密 -> 分卷上传; 下载逐卷拉取本地解密。 + 服务端无密钥不解密, 文件名也加密 (enc: URL-safe base64)。 + . + 包含: + * sz-transfer CLI (上传/下载/列表/删除/配额/账号) + * sz-tui 终端界面 (登录门禁, 12 项菜单) + * sz-config 配置 (服务器/token/密钥) + . + 配置存 ~/.config/7z-encrypt/, 密钥存 ~/.local/share/7z-encrypt/。 diff --git a/debian/copyright b/debian/copyright new file mode 100644 index 0000000..7d292b7 --- /dev/null +++ b/debian/copyright @@ -0,0 +1,27 @@ +Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/ +Upstream-Name: 7z-encrypt +Upstream-Contact: edgevoid +Source: https://gitee.com/edgevoid/7z-encrypt + +Files: * +Copyright: 2026 edgevoid +License: MIT + +License: MIT + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + . + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + . + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE. diff --git a/debian/debhelper-build-stamp b/debian/debhelper-build-stamp new file mode 100644 index 0000000..fa19433 --- /dev/null +++ b/debian/debhelper-build-stamp @@ -0,0 +1 @@ +7z-encrypt-client diff --git a/debian/files b/debian/files new file mode 100644 index 0000000..a881b4a --- /dev/null +++ b/debian/files @@ -0,0 +1,2 @@ +7z-encrypt-client_1.0.0_all.deb utils optional architecture=all +7z-encrypt-client_1.0.0_amd64.buildinfo utils optional diff --git a/debian/postinst b/debian/postinst new file mode 100755 index 0000000..1597b0a --- /dev/null +++ b/debian/postinst @@ -0,0 +1,16 @@ +#!/bin/sh +set -e + +case "$1" in + configure) + P=/usr/share/7z-encrypt/.venv + if [ ! -x "$P/bin/python" ]; then + echo "正在创建 7z-encrypt 运行环境 (venv + cryptography/tqdm) ..." + python3 -m venv "$P" + "$P/bin/pip" install --quiet --upgrade pip || true + "$P/bin/pip" install --quiet cryptography tqdm + fi + ;; +esac + +exit 0 diff --git a/debian/prerm b/debian/prerm new file mode 100755 index 0000000..016f0f1 --- /dev/null +++ b/debian/prerm @@ -0,0 +1,8 @@ +#!/bin/sh +set -e + +if [ "$1" = "remove" ] || [ "$1" = "purge" ]; then + rm -rf /usr/share/7z-encrypt/.venv 2>/dev/null || true +fi + +exit 0 diff --git a/debian/rules b/debian/rules new file mode 100755 index 0000000..9f83f7f --- /dev/null +++ b/debian/rules @@ -0,0 +1,25 @@ +#!/usr/bin/make -f +%: + dh $@ + +override_dh_auto_build: + +override_dh_auto_test: + +override_dh_auto_install: + # 客户端源码 + mkdir -p debian/7z-encrypt-client/usr/share/7z-encrypt + cp transfer.py metadata.py crypto.py splitter.py config.py state.py \ + debian/7z-encrypt-client/usr/share/7z-encrypt/ + cp ../tui/scripts/tui.py debian/7z-encrypt-client/usr/share/7z-encrypt/ + # CLI wrapper + mkdir -p debian/7z-encrypt-client/usr/bin + printf '#!/bin/sh\nexec /usr/share/7z-encrypt/.venv/bin/python /usr/share/7z-encrypt/transfer.py "$$@"\n' \ + > debian/7z-encrypt-client/usr/bin/sz-transfer + printf '#!/bin/sh\nexport SZ_CLIENT_DIR=/usr/share/7z-encrypt\nexec /usr/share/7z-encrypt/.venv/bin/python /usr/share/7z-encrypt/tui.py "$$@"\n' \ + > debian/7z-encrypt-client/usr/bin/sz-tui + printf '#!/bin/sh\nexec /usr/share/7z-encrypt/.venv/bin/python /usr/share/7z-encrypt/config.py "$$@"\n' \ + > debian/7z-encrypt-client/usr/bin/sz-config + chmod 755 debian/7z-encrypt-client/usr/bin/sz-transfer \ + debian/7z-encrypt-client/usr/bin/sz-tui \ + debian/7z-encrypt-client/usr/bin/sz-config diff --git a/debian/source/format b/debian/source/format new file mode 100644 index 0000000..89ae9db --- /dev/null +++ b/debian/source/format @@ -0,0 +1 @@ +3.0 (native) diff --git a/transfer.py b/transfer.py index 5c9d646..80710fd 100644 --- a/transfer.py +++ b/transfer.py @@ -16,6 +16,7 @@ import argparse import base64 import json +import os import shutil import subprocess import sys @@ -326,27 +327,54 @@ def _parse_disposition(cd: str) -> str: return _safe_name(plain if plain is not None else name) -def _load_token() -> str: - """从 config/config.json 读 server.token (CLI 默认来源)""" +def _config_path() -> Path: + """配置文件路径: SZ_CONFIG env -> XDG (~/.config/7z-encrypt/config.json) + -> 兼容旧式相对 config/config.json (源码开发模式) + + 安装版 (deb) cwd 不可写, 配置必须走用户级 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("config/config.json") + return xdg_cfg if not legacy.exists() else legacy + + +def _config_server() -> str: + """从配置文件读 server.url (安装版 CLI 默认服务器)""" try: - cfg = Path("config/config.json") + cfg = _config_path() + return str(json.loads(cfg.read_text(encoding="utf-8")).get("server", {}).get("url", "")) + except (OSError, json.JSONDecodeError): + return "" + + +def _load_token() -> str: + """从配置文件读 server.token (CLI 默认来源, XDG 优先)""" + try: + cfg = _config_path() return str(json.loads(cfg.read_text(encoding="utf-8")).get("server", {}).get("token", "")) except (OSError, json.JSONDecodeError): return "" def _save_token(token: str) -> None: - """登录/注册成功后 token 持久化到 config/config.json server.token""" - cfg = Path("config/config.json") + """登录/注册成功后 token 持久化到配置文件 server.token""" + cfg = _config_path() + cfg.parent.mkdir(parents=True, exist_ok=True) data = json.loads(cfg.read_text(encoding="utf-8")) if cfg.exists() else {} data.setdefault("server", {})["token"] = token cfg.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def _load_keyring_path() -> Path: - """密钥库路径: config.json 显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)""" + """密钥库路径: 配置显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)""" try: - cfg = Path("config/config.json") + cfg = _config_path() raw = str(json.loads(cfg.read_text(encoding="utf-8")).get("crypto", {}).get("keyring_path", "")) if raw: return Path(raw) @@ -358,7 +386,7 @@ def _load_keyring_path() -> Path: def main() -> int: ap = argparse.ArgumentParser(description="传输客户端: 上传 / 文件列表 / 下载") - ap.add_argument("--server", required=True, help="服务端地址, 如 http://127.0.0.1:8000") + ap.add_argument("--server", default=None, help="服务端地址 (默认读配置 server.url)") ap.add_argument("--init", help="init json 路径 (上传模式)") ap.add_argument("--manifest", help="卷清单路径 (上传模式)") ap.add_argument("--list", action="store_true", help="列出服务端文件") @@ -374,8 +402,11 @@ def main() -> int: ap.add_argument("--whoami", action="store_true", help="显示当前登录用户") args = ap.parse_args() + server = args.server or _config_server() + if not server: + ap.error("未指定 --server 且配置无 server.url (先 sz-config --set-server)") token = args.token if args.token is not None else _load_token() - client = TransferClient(args.server, token=token) + client = TransferClient(server, token=token) if args.whoami: try: