From 34496b4bf728400803b56558765a98de18e982a8 Mon Sep 17 00:00:00 2001 From: lou Date: Mon, 10 Aug 2026 01:30:19 +0800 Subject: [PATCH] =?UTF-8?q?=E9=9B=B6=E5=90=88=E5=B9=B6=E9=87=8D=E6=9E=84:?= =?UTF-8?q?=20complete=20=E5=8F=AA=E6=A0=A1=E9=AA=8C=E5=8D=B7=E9=BD=90+?= =?UTF-8?q?=E5=8D=B7=E7=9B=AE=E5=BD=95=20move=20=E5=85=A5=E5=BA=93=20(?= =?UTF-8?q?=E7=A7=92=E5=9B=9E);=20=E4=B8=8B=E8=BD=BD=E6=94=B9=E9=80=90?= =?UTF-8?q?=E5=8D=B7=20GET=20chunk/{n};=20=E5=88=A0=20assembler;=20?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E6=9C=BA=20uploading->storing->done;=20?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=207=20=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api.py | 75 +++++++++++++++++++++++------------------------ assembler.py | 55 ---------------------------------- storage.py | 32 +++++++++----------- task_manager.py | 6 ++-- tests/test_api.py | 35 +++++++++++++++------- 5 files changed, 78 insertions(+), 125 deletions(-) delete mode 100644 assembler.py diff --git a/api.py b/api.py index 02b9aa0..e1c16e2 100644 --- a/api.py +++ b/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 (与客户端预共享) """ 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: - h = hashlib.sha256() - with open(path, "rb") as f: - for chunk in iter(lambda: f.read(1 << 20), b""): - h.update(chunk) - return h.hexdigest() +def _enc_headers(rec: dict) -> dict[str, str]: + """下载响应头: 解密参数 + 卷数""" + ep = json.dumps(rec["enc_params"], ensure_ascii=False) + return { + "X-Enc-Params": base64.b64encode(ep.encode()).decode(), + "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)]) -async def download_file(file_id: str): - """下载密文文件。enc_params 放响应头 X-Enc-Params (客户端本地解密用)""" +@app.get("/api/files/{file_id}/chunk/{idx}", dependencies=[Depends(verify_token)]) +async def download_chunk(file_id: str, idx: int): + """下载单卷密文 (零合并: 服务端不拼接, 客户端逐卷拉取本地合并)""" rec = db.get_file(file_id) if rec is None: return JSONResponse(status_code=404, content={"error": "文件不存在"}) - path = Path(rec["path"]) - if not path.exists(): - return JSONResponse(status_code=404, content={"error": f"存储文件缺失: {path}"}) - # 解密参数: 任务 init 时存的 enc {alg, key_id, context} - enc_params = json.dumps(rec["enc_params"], ensure_ascii=False) - headers = { - "X-Enc-Params": base64.b64encode(enc_params.encode()).decode(), - "X-Chunk-Count": str(rec["chunk_count"]), # 客户端按卷粒度显示进度 - } + if not 1 <= idx <= rec["chunk_count"]: + return JSONResponse(status_code=404, content={"error": f"卷号越界: {idx}"}) + chunk_path = Path(rec["path"]) / f"chunk_{idx:04d}" + if not chunk_path.exists(): + return JSONResponse(status_code=404, content={"error": f"卷文件缺失: {chunk_path}"}) 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 - # 零知识: 合并后的密文直接入库, 服务端不碰解密 - final_path = storage.store(merged, transfer["file_name"]) + # 零合并: 只校验卷齐, 卷目录直接 move 入存储区, 秒回 + 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), - final_path.stat().st_size, - _sha256(final_path), + str(final_dir), + transfer["file_size"], + 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"}) diff --git a/assembler.py b/assembler.py deleted file mode 100644 index 7431e07..0000000 --- a/assembler.py +++ /dev/null @@ -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 diff --git a/storage.py b/storage.py index e666339..1016237 100644 --- a/storage.py +++ b/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) - - # 同名冲突: 加 -1/-2 后缀 - 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 + shutil.move(str(chunk_dir), str(final_dir)) + return final_dir diff --git a/task_manager.py b/task_manager.py index c55226f..f10f7e3 100644 --- a/task_manager.py +++ b/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_DECRYPTING = "decrypting" +STATUS_STORING = "storing" STATUS_DONE = "done" STATUS_FAILED = "failed" diff --git a/tests/test_api.py b/tests/test_api.py index 4358234..8387322 100644 --- a/tests/test_api.py +++ b/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: - stored = f.read() + chunk_dir = Path(row["path"]) + 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, 不新增入库记录