账号体系: 注册/登录/注销 CLI + rotate_key 轮换密钥 + config.py --regenerate-key(--yes); TUI 菜单 12 项

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 解析、注册/登录/注销、用户文件隔离。
 
 ---
 

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},
                 },

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:

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()