零合并重构: complete 只校验卷齐+卷目录 move 入库 (秒回); 下载改逐卷 GET chunk/{n}; 删 assembler; 状态机 uploading->storing->done; 测试 7 项

api.py

@@ -4,26 +4,24 @@ API 层 (api)
 输入: HTTP 请求 (带认证)
 输出: 响应 / 路由到各模块
 
-零知识设计: 服务端只传输+存储密文, 不持有密钥, 不解密。
+零知识 + 零合并设计: 服务端只存卷/传卷, 不合并不解密, 不持有密钥。
 端点:
     POST /api/transfer/init         {init json} -> {transfer_id}
     GET  /api/transfer/{id}/chunks  -> {received: [n...]}
     PUT  /api/transfer/{id}/chunk/{n}  卷密文 -> {ok} / 409
-    POST /api/transfer/{id}/complete    -> {status, file_id} / 409
+    POST /api/transfer/{id}/complete    -> {status, file_id} / 409 (秒回, 不合并)
     GET  /api/files                  -> {files: [...]}
-    GET  /api/files/{file_id}        -> 密文文件 (头 X-Enc-Params: base64(enc_params))
+    GET  /api/files/{id}/chunk/{n}   -> 单卷密文 (头 X-Enc-Params / X-Chunk-Count)
 全部端点需 Authorization: Bearer <token> (与客户端预共享)
 """
 
 import base64
-import hashlib
 import json
 from pathlib import Path
 
 from fastapi import Depends, FastAPI, Header, HTTPException, Request
 from fastapi.responses import FileResponse, JSONResponse
 
-from assembler import Assembler, AssemblerError
 from db import ServerDB
 from receiver import Receiver, ReceiverError
 from settings import STORAGE_ROOT, TMP_ROOT, TOKEN
@@ -34,10 +32,9 @@ from task_manager import TaskManager
 db = ServerDB()
 tasks = TaskManager(db)
 receiver = Receiver(tasks, TMP_ROOT)
-assembler = Assembler(tasks, TMP_ROOT)
 storage = Storage(tasks, STORAGE_ROOT)
 
-app = FastAPI(title="7z-encrypt 服务端 (零知识)")
+app = FastAPI(title="7z-encrypt 服务端 (零知识 + 零合并)")
 
 
 # ---------- 认证 ----------
@@ -48,15 +45,16 @@ def verify_token(authorization: str | None = Header(default=None)) -> None:
         raise HTTPException(status_code=401, detail="未授权: token 无效或缺失")
 
 
-def _sha256(path: str | Path) -> str:
+def _enc_headers(rec: dict) -> dict[str, str]:
-    h = hashlib.sha256()
+    """下载响应头: 解密参数 + 卷数"""
-    with open(path, "rb") as f:
+    ep = json.dumps(rec["enc_params"], ensure_ascii=False)
-        for chunk in iter(lambda: f.read(1 << 20), b""):
+    return {
-            h.update(chunk)
+        "X-Enc-Params": base64.b64encode(ep.encode()).decode(),
-    return h.hexdigest()
+        "X-Chunk-Count": str(rec["chunk_count"]),
+    }
 
 
-# ---------- 文件: ls / 下载 (密文 + 解密参数) ----------
+# ---------- 文件: ls / 逐卷下载 ----------
 
 @app.get("/api/files", dependencies=[Depends(verify_token)])
 async def list_files():
@@ -64,26 +62,22 @@ async def list_files():
     return {"files": db.list_files()}
 
 
-@app.get("/api/files/{file_id}", dependencies=[Depends(verify_token)])
+@app.get("/api/files/{file_id}/chunk/{idx}", dependencies=[Depends(verify_token)])
-async def download_file(file_id: str):
+async def download_chunk(file_id: str, idx: int):
-    """下载密文文件。enc_params 放响应头 X-Enc-Params (客户端本地解密用)"""
+    """下载单卷密文 (零合并: 服务端不拼接, 客户端逐卷拉取本地合并)"""
     rec = db.get_file(file_id)
     if rec is None:
         return JSONResponse(status_code=404, content={"error": "文件不存在"})
-    path = Path(rec["path"])
+    if not 1 <= idx <= rec["chunk_count"]:
-    if not path.exists():
+        return JSONResponse(status_code=404, content={"error": f"卷号越界: {idx}"})
-        return JSONResponse(status_code=404, content={"error": f"存储文件缺失: {path}"})
+    chunk_path = Path(rec["path"]) / f"chunk_{idx:04d}"
-    # 解密参数: 任务 init 时存的 enc {alg, key_id, context}
+    if not chunk_path.exists():
-    enc_params = json.dumps(rec["enc_params"], ensure_ascii=False)
+        return JSONResponse(status_code=404, content={"error": f"卷文件缺失: {chunk_path}"})
-    headers = {
-        "X-Enc-Params": base64.b64encode(enc_params.encode()).decode(),
-        "X-Chunk-Count": str(rec["chunk_count"]),  # 客户端按卷粒度显示进度
-    }
     return FileResponse(
-        path,
+        chunk_path,
         filename=rec["file_name"],
         media_type="application/octet-stream",
-        headers=headers,
+        headers=_enc_headers(rec),
     )
 
 
@@ -122,27 +116,32 @@ async def put_chunk(transfer_id: str, idx: int, request: Request):
 async def complete(transfer_id: str):
     if tasks.get(transfer_id) is None:
         return JSONResponse(status_code=404, content={"error": "任务不存在"})
-    # 幂等: 已完成的任务直接返回已有 file_id (客户端超时重试/断线重连场景)
+    # 幂等: 已完成的任务直接返回已有 file_id
     done_rec = db.get_file_by_transfer(transfer_id)
     if done_rec is not None:
         return {"status": "done", "file_id": done_rec["file_id"]}
     try:
-        tasks.set_status(transfer_id, "assembling")
-        merged = assembler.assemble(transfer_id)
-
         transfer = tasks.get(transfer_id)
         assert transfer is not None
-        # 零知识: 合并后的密文直接入库, 服务端不碰解密
+        # 零合并: 只校验卷齐, 卷目录直接 move 入存储区, 秒回
-        final_path = storage.store(merged, transfer["file_name"])
+        got = tasks.received(transfer_id)
+        if len(got) != transfer["chunk_count"]:
+            return JSONResponse(status_code=409, content={
+                "status": "incomplete",
+                "received": sorted(got),
+                "error": f"缺卷: {transfer['chunk_count'] - len(got)} 卷未上传",
+            })
+        tasks.set_status(transfer_id, "storing")
+        final_dir = storage.store_chunks(transfer_id, TMP_ROOT / transfer_id)
         file_id = db.insert_file(
             transfer_id,
             transfer["file_name"],
-            str(final_path),
+            str(final_dir),
-            final_path.stat().st_size,
+            transfer["file_size"],
-            _sha256(final_path),
+            transfer["total_sha256"],
         )
         tasks.set_status(transfer_id, "done")
         return {"status": "done", "file_id": file_id}
-    except AssemblerError as e:
+    except OSError as e:
         tasks.set_status(transfer_id, "failed")
         return JSONResponse(status_code=409, content={"error": str(e), "status": "failed"})

assembler.py

@@ -1,55 +0,0 @@
-"""
-汇聚器模块 (assembler)
-模块: 服务端 / 汇聚器
-输入: 已收卷集合
-输出: 完整密文文件
-
-两道门: 卷齐才合并 (缺卷返回列表让客户端补); 合并后总哈希必须匹配才进解包。
-"""
-
-import hashlib
-import shutil
-from pathlib import Path
-
-from task_manager import TaskManager
-
-
-class AssemblerError(RuntimeError):
-    """汇聚失败 (卷未齐/总哈希不一致)"""
-
-
-class Assembler:
-    def __init__(self, tasks: TaskManager, tmp_root: str | Path) -> None:
-        self.tasks = tasks
-        self.tmp_root = Path(tmp_root)
-
-    def assemble(self, transfer_id: str) -> Path:
-        """按 idx 顺序合并全部卷 -> 返回合并后密文路径"""
-        transfer = self.tasks.get(transfer_id)
-        if transfer is None:
-            raise AssemblerError(f"任务不存在: {transfer_id}")
-        total = transfer["chunk_count"]
-
-        received = self.tasks.received(transfer_id)
-        if len(received) < total:
-            missing = sorted(set(range(1, total + 1)) - received)
-            raise AssemblerError(f"缺卷 {missing}, 等齐再合并")
-
-        chunk_dir = self.tmp_root / transfer_id
-        merged_path = chunk_dir / "merged.bin"
-        with open(merged_path, "wb") as out:
-            for idx in range(1, total + 1):
-                chunk_path = chunk_dir / f"chunk_{idx:04d}"
-                with open(chunk_path, "rb") as f:
-                    shutil.copyfileobj(f, out)
-
-        # 总哈希校验
-        h = hashlib.sha256()
-        with open(merged_path, "rb") as f:
-            for chunk in iter(lambda: f.read(1 << 20), b""):
-                h.update(chunk)
-        if h.hexdigest() != transfer["total_sha256"]:
-            merged_path.unlink(missing_ok=True)
-            raise AssemblerError("合并后总 SHA-256 不一致, 任务 failed (需整体重传)")
-
-        return merged_path

storage.py

@@ -1,11 +1,11 @@
 """
 存储层模块 (storage)
 模块: 服务端 / 存储层
-输入: 合并后的密文文件 + 元数据
+输入: 已校验的密文卷目录 + 元数据
-输出: 最终文件路径
+输出: 最终存储目录
 
-零知识: 只存密文, 服务端无密钥。
+零知识 + 零合并: 只存卷, 不做任何拼接/合并, 服务端无密钥。
-按日期目录落盘, 同名冲突加后缀去重, 流式移动不整块进内存。
+卷目录整体 move 到日期/transfer_id 目录 (transfer_id 唯一, 无同名冲突)。
 """
 
 import shutil
@@ -20,19 +20,15 @@ class Storage:
         self.tasks = tasks
         self.storage_root = Path(storage_root)
 
-    def store(self, enc_path: Path, file_name: str) -> Path:
+    def store_chunks(self, transfer_id: str, chunk_dir: Path) -> Path:
-        """密文文件 -> 最终存储路径 (返回路径)"""
+        """卷目录 -> 最终存储目录 (整体 move, 返回目标目录路径)
+
+        注意: 目标目录不能预创建, 否则 shutil.move 会嵌套成 dst/dst/。
+        """
         day_dir = self.storage_root / datetime.now().strftime("%Y-%m-%d")
+        final_dir = day_dir / transfer_id
+        if final_dir.exists():
+            raise OSError(f"存储目录已存在: {final_dir}")
         day_dir.mkdir(parents=True, exist_ok=True)
-
+        shutil.move(str(chunk_dir), str(final_dir))
-        # 同名冲突: 加 -1/-2 后缀
+        return final_dir
-        target = day_dir / file_name
-        counter = 1
-        while target.exists():
-            stem, suffix = file_name.rsplit(".", 1) if "." in file_name else (file_name, "")
-            name = f"{stem}-{counter}" + (f".{suffix}" if suffix else "")
-            target = day_dir / name
-            counter += 1
-
-        shutil.move(str(enc_path), str(target))
-        return target

task_manager.py

@@ -4,7 +4,8 @@
 输入: init 请求 / 卷到达事件
 输出: transfer 状态机
 
-状态机: uploading -> assembling -> decrypting -> done / failed
+状态机: uploading -> storing -> done / failed
+(零合并: 无 assembling/decrypting 阶段, 卷齐即入库)
 transfer_id 生命周期贯穿接收->入库; 孤儿任务由清理逻辑回收 (见 server/data/tmp)。
 """
 
@@ -14,8 +15,7 @@ from typing import Any
 from db import ServerDB
 
 STATUS_UPLOADING = "uploading"
-STATUS_ASSEMBLING = "assembling"
+STATUS_STORING = "storing"
-STATUS_DECRYPTING = "decrypting"
 STATUS_DONE = "done"
 STATUS_FAILED = "failed"
 

tests/test_api.py

@@ -95,7 +95,7 @@ class ServerPipelineTest(unittest.TestCase):
         bad = {"Authorization": "Bearer wrong"}
         self.assertEqual(self.client.get("/api/files", headers=bad).status_code, 401)
 
-    # ---------- 全链路 (零知识: 服务端只存密文) ----------
+    # ---------- 全链路 (零知识 + 零合并: 服务端只存卷) ----------
 
     def test_full_pipeline_roundtrip(self):
         data = os.urandom(3 << 20)  # 3MB
@@ -108,17 +108,20 @@ class ServerPipelineTest(unittest.TestCase):
         file_id = r.json()["file_id"]
         self.assertTrue(file_id)
 
-        # 存储的是密文 (不是明文 data), 零知识验证
+        # 零合并: 存储的是卷目录 (chunk_XXXX 文件), 不是合并单文件, 没有明文
         db = ServerDB()
         with db._conn() as conn:
             row = conn.execute(
-                "SELECT path, size FROM files WHERE file_id = ?", (file_id,)
+                "SELECT path FROM files WHERE file_id = ?", (file_id,)
             ).fetchone()
         self.assertIsNotNone(row)
-        with open(row["path"], "rb") as f:
+        chunk_dir = Path(row["path"])
-            stored = f.read()
+        self.assertTrue(chunk_dir.is_dir())
+        chunk_files_on_disk = sorted(chunk_dir.glob("chunk_*"))
+        self.assertEqual(len(chunk_files_on_disk), len(chunk_files))
+        # 卷内容与客户端上传的卷一致
+        stored = b"".join(p.read_bytes() for p in chunk_files_on_disk)
         self.assertEqual(stored, enc_bytes)
-        self.assertEqual(row["size"], len(enc_bytes))
         self.assertNotEqual(stored, data)  # 服务端手里不是明文
 
         # transfer 状态 done
@@ -200,19 +203,29 @@ class ServerPipelineTest(unittest.TestCase):
         files = r.json()["files"]
         self.assertTrue(any(f["file_id"] == file_id for f in files))
 
-        # 下载返回密文 + 解密参数头 (客户端本地解密用)
+        # 逐卷下载: 每卷内容与上传卷一致, 拼接 == 完整密文
-        r = self.client.get(f"/api/files/{file_id}", headers=AUTH)
+        r = self.client.get(f"/api/files/{file_id}/chunk/1", headers=AUTH)
         self.assertEqual(r.status_code, 200, r.text)
-        self.assertEqual(r.content, enc_bytes)
         enc_params = json.loads(
             __import__("base64").b64decode(r.headers["X-Enc-Params"]).decode()
         )
         self.assertEqual(enc_params["key_id"], TEST_KEY_ID)
         self.assertEqual(enc_params["alg"], "cobblestone-aes256gcm")
-
+        self.assertEqual(r.headers["X-Chunk-Count"], str(len(chunk_files)))
+        # 逐卷拉取拼接 == 完整密文
+        pieces = []
+        for idx in range(1, len(chunk_files) + 1):
+            r = self.client.get(f"/api/files/{file_id}/chunk/{idx}", headers=AUTH)
+            self.assertEqual(r.status_code, 200, r.text)
+            pieces.append(r.content)
+        self.assertEqual(b"".join(pieces), enc_bytes)
+        # 越界卷 -> 404
+        self.assertEqual(
+            self.client.get(f"/api/files/{file_id}/chunk/999", headers=AUTH).status_code, 404
+        )
         # 未知 file_id -> 404
         self.assertEqual(
-            self.client.get("/api/files/nope", headers=AUTH).status_code, 404
+            self.client.get("/api/files/nope/chunk/1", headers=AUTH).status_code, 404
         )
 
         # complete 幂等: 重复 complete 返回同一个 file_id, 不新增入库记录