From 214c943c077d37ce7b8e773b06222b9d0a68d54c Mon Sep 17 00:00:00 2001 From: lou Date: Mon, 10 Aug 2026 15:27:43 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B4=A6=E5=8F=B7=E4=BD=93=E7=B3=BB:=20?= =?UTF-8?q?=E6=B3=A8=E5=86=8C/=E7=99=BB=E5=BD=95/=E6=B3=A8=E9=94=80=20CLI?= =?UTF-8?q?=20+=20rotate=5Fkey=20=E8=BD=AE=E6=8D=A2=E5=AF=86=E9=92=A5=20+?= =?UTF-8?q?=20config.py=20--regenerate-key(--yes);=20TUI=20=E8=8F=9C?= =?UTF-8?q?=E5=8D=95=2012=20=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 14 ++++++++--- config.py | 34 ++++++++++++++++++++++++++- crypto.py | 10 +++++++- transfer.py | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 120 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 1de3b59..33d7a75 100644 --- a/README.md +++ b/README.md @@ -87,8 +87,16 @@ - 密钥默认存系统标准位置 `~/.local/share/7z-encrypt/keyring.json`（XDG），环境变量 `SZ_KEYRING` 可指定；旧版位置自动迁移 - 密钥导出/导入（TUI 菜单 7/8）：红色风险警告 + 二次确认（yes/CONFIRM），导入自动备份 `keyring.json.bak`，导出文件权限 600 +- 重新生成密钥（TUI 菜单 12）：红色警告 + 二次确认，旧密文永久不可解（先导出备份） - 换机迁移：旧机导出 → 新机导入 → 直接解密旧密文 +### 11. 账号体系 + +- 注册 / 登录 / 注销（TUI 菜单 9/10/11 或 CLI `--register/--login/--logout`） +- 密码 pbkdf2-hmac-sha256 哈希存储（20 万轮迭代，不落明文），登录 token 随机 32 字节 +- 文件按用户隔离：注册用户只见自己的文件，admin（SZ_TOKEN）可见全部 +- 登录后 token 自动保存到 config.json，注销即失效 + --- ## 与百度网盘对比 @@ -133,17 +141,17 @@ python config.py --set-token 你的token # 3. TUI (推荐) python3 ../tui/scripts/tui.py # 菜单: 1上传 / 2任务续传 / 3配置 / 4文件列表 / 5下载 / 6删除 -# 7密钥导出 / 8密钥导入 (红色警告+二次确认) / q退出 +# 7密钥导出 / 8密钥导入 / 9注册 / 10登录 / 11注销 / 12重新生成密钥 / q退出 ``` ## 测试 ```bash cd 7z-encrypt && .venv/bin/pytest -q # 71 项 -cd server && .venv/bin/pytest -q # 9 项 +cd server && .venv/bin/pytest -q # 12 项 ``` -覆盖：加密/解密回环、7z 压缩回环、逐卷上传下载、断点续传、token 认证、文件名加密回环、路径穿越防护、配额、幂等、坏卷拒绝、超大清单拒绝、密钥路径 env/XDG 解析。 +覆盖：加密/解密回环、7z 压缩回环、逐卷上传下载、断点续传、token 认证、文件名加密回环、路径穿越防护、配额、幂等、坏卷拒绝、超大清单拒绝、密钥路径 env/XDG 解析、注册/登录/注销、用户文件隔离。 --- diff --git a/config.py b/config.py index 558569c..f160e04 100644 --- a/config.py +++ b/config.py @@ -152,8 +152,40 @@ def main() -> int: 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(): @@ -170,7 +202,7 @@ def main() -> int: "max_retries": 3, "retry_delay": 1.0, }, - "crypto": {"key_id": "default_key", "keyring_path": "config/keyring.json"}, + "crypto": {"key_id": "default_key", "keyring_path": ""}, "splitter": {"chunk_size_mb": 10}, "state": {"db_path": "config/tasks.db", "cleanup_days": 7}, }, diff --git a/crypto.py b/crypto.py index 38cd471..a1566db 100644 --- a/crypto.py +++ b/crypto.py @@ -103,12 +103,20 @@ class CryptoEngine: def generate_key(self, key_id: str) -> bytes: """生成新密钥 (os.urandom 32B) 并持久化。已存在则不覆盖, 抛错防误用""" if key_id in self._keyring: - raise CryptoError(f"密钥ID '{key_id}' 已存在, 不覆盖 (需轮换请换新 key_id)") + 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: diff --git a/transfer.py b/transfer.py index fc8f4ff..0ea6c57 100644 --- a/transfer.py +++ b/transfer.py @@ -84,6 +84,32 @@ class TransferClient: # ---------- 协议端点 ---------- + 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( @@ -302,6 +328,14 @@ def _load_token() -> str: return "" +def _save_token(token: str) -> None: + """登录/注册成功后 token 持久化到 config/config.json server.token""" + cfg = Path("config/config.json") + 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)""" try: @@ -327,11 +361,44 @@ def main() -> int: 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 失效)") args = ap.parse_args() token = args.token if args.token is not None else _load_token() client = TransferClient(args.server, token=token) + 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()