commit 28654c0056a1fb7161325cd06a813a4aff44cb62 Author: lou Date: Sun Aug 9 23:24:03 2026 +0800 服务端 9 模块骨架: api/receiver/assembler/unpacker/decrypt/storage/db/keyring/task_manager diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f960d3e --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +# 虚拟环境 +.venv/ +__pycache__/ +*.pyc + +# 运行时数据 (分卷/合并/解密中间产物 + 明文存储 + 密钥) +data/ diff --git a/api.py b/api.py new file mode 100644 index 0000000..43e26dc --- /dev/null +++ b/api.py @@ -0,0 +1,106 @@ +""" +API 层 (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 +""" + +import hashlib + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from assembler import Assembler, AssemblerError +from settings import KEYRING_PATH, STORAGE_ROOT, TMP_ROOT +from db import ServerDB +from decrypt import DecryptError, Decryptor +from keyring import Keyring +from receiver import Receiver, ReceiverError +from storage import Storage +from task_manager import TaskManager +from unpacker import UnpackError, Unpacker + +# ---------- 依赖装配 (单例) ---------- +db = ServerDB() +tasks = TaskManager(db) +keyring = Keyring(KEYRING_PATH) +receiver = Receiver(tasks, TMP_ROOT) +assembler = Assembler(tasks, TMP_ROOT) +unpacker = Unpacker(tasks) +decryptor = Decryptor(keyring) +storage = Storage(tasks, STORAGE_ROOT) + +app = FastAPI(title="7z-encrypt 服务端") + + +def _sha256(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() + + +@app.post("/api/transfer/init") +async def init_transfer(request: Request): + try: + init_json = await request.json() + transfer_id = tasks.create(init_json) + except (KeyError, TypeError, ValueError) as e: + return JSONResponse(status_code=400, content={"error": f"init json 非法: {e}"}) + return {"transfer_id": transfer_id} + + +@app.get("/api/transfer/{transfer_id}/chunks") +async def get_chunks(transfer_id: str): + if tasks.get(transfer_id) is None: + return JSONResponse(status_code=404, content={"error": "任务不存在"}) + return {"received": sorted(tasks.received(transfer_id))} + + +@app.put("/api/transfer/{transfer_id}/chunk/{idx}") +async def put_chunk(transfer_id: str, idx: int, request: Request): + if tasks.get(transfer_id) is None: + return JSONResponse(status_code=404, content={"error": "任务不存在"}) + data = await request.body() + try: + receiver.receive(transfer_id, idx, data) + except ReceiverError as e: + return JSONResponse(status_code=409, content={"error": str(e)}) + return {"ok": True} + + +@app.post("/api/transfer/{transfer_id}/complete") +async def complete(transfer_id: str): + if tasks.get(transfer_id) is None: + return JSONResponse(status_code=404, content={"error": "任务不存在"}) + try: + tasks.set_status(transfer_id, "assembling") + merged = assembler.assemble(transfer_id) + + tasks.set_status(transfer_id, "decrypting") + merged_path, enc_params = unpacker.unpack(transfer_id, merged) + plain_path = TMP_ROOT / transfer_id / "plain.bin" + decryptor.decrypt(merged_path, enc_params, plain_path) + + transfer = tasks.get(transfer_id) + assert transfer is not None + final_path = storage.store(plain_path, transfer["file_name"]) + file_id = db.insert_file( + transfer_id, + transfer["file_name"], + str(final_path), + final_path.stat().st_size, + _sha256(final_path), + ) + tasks.set_status(transfer_id, "done") + return {"status": "done", "file_id": file_id} + except (AssemblerError, UnpackError, DecryptError) 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 new file mode 100644 index 0000000..7431e07 --- /dev/null +++ b/assembler.py @@ -0,0 +1,55 @@ +""" +汇聚器模块 (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/db.py b/db.py new file mode 100644 index 0000000..b574830 --- /dev/null +++ b/db.py @@ -0,0 +1,170 @@ +""" +入库层 (db) +模块: 服务端 / 入库层 +输入: 文件路径 + 元数据 +输出: file_id + +PostgreSQL 表: + transfers 任务主表 (状态机 uploading->assembling->decrypting->done/failed) + transfer_chunks 每卷信息 + 接收时间 (null = 未收, 断点续传依据) + files 明文文件记录 +""" + +import uuid +from typing import Any, cast + +import psycopg +from psycopg.rows import dict_row +from psycopg.types.json import Jsonb + +from settings import DATABASE_URL + + +class ServerDB: + """服务端数据库访问 (psycopg3, 每次操作独立连接)""" + + def __init__(self, url: str = DATABASE_URL) -> None: + self.url = url + self._init_schema() + + def _conn(self): + return psycopg.connect(self.url, row_factory=cast(Any, dict_row)) + + def _init_schema(self) -> None: + with self._conn() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS transfers ( + transfer_id TEXT PRIMARY KEY, + file_name TEXT NOT NULL, + file_size BIGINT NOT NULL, + chunk_count INTEGER NOT NULL, + total_sha256 TEXT NOT NULL, + enc_params JSONB NOT NULL, + status TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS transfer_chunks ( + transfer_id TEXT NOT NULL REFERENCES transfers(transfer_id), + idx INTEGER NOT NULL, + size BIGINT NOT NULL, + sha256 TEXT NOT NULL, + received_at TIMESTAMPTZ, + PRIMARY KEY (transfer_id, idx) + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS files ( + file_id TEXT PRIMARY KEY, + transfer_id TEXT NOT NULL REFERENCES transfers(transfer_id), + file_name TEXT NOT NULL, + path TEXT NOT NULL, + size BIGINT NOT NULL, + sha256 TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """ + ) + + # ---------- transfers ---------- + + def create_transfer(self, transfer_id: str, init_json: dict[str, Any]) -> None: + enc = init_json["enc"] + with self._conn() as conn: + conn.execute( + "INSERT INTO transfers (transfer_id, file_name, file_size, chunk_count, total_sha256, enc_params, status) " + "VALUES (%s, %s, %s, %s, %s, %s, %s)", + ( + transfer_id, + init_json["file_name"], + init_json["file_size"], + init_json["chunk_count"], + init_json["total_sha256"], + Jsonb(enc), + "uploading", + ), + ) + conn.executemany( + "INSERT INTO transfer_chunks (transfer_id, idx, size, sha256) VALUES (%s, %s, %s, %s)", + [ + (transfer_id, ch["index"], ch["size"], ch["sha256"]) + for ch in init_json["chunks"] + ], + ) + + def get_transfer(self, transfer_id: str) -> dict[str, Any] | None: + with self._conn() as conn: + row = conn.execute( + "SELECT * FROM transfers WHERE transfer_id = %s", (transfer_id,) + ).fetchone() + if not row: + return None + return dict(row) + + def set_status(self, transfer_id: str, status: str) -> None: + with self._conn() as conn: + conn.execute( + "UPDATE transfers SET status = %s, updated_at = now() WHERE transfer_id = %s", + (status, transfer_id), + ) + + # ---------- chunks ---------- + + def get_chunk_spec(self, transfer_id: str, idx: int) -> dict[str, Any] | None: + """卷规格 (size/sha256), 接收校验用""" + with self._conn() as conn: + row = conn.execute( + "SELECT size, sha256 FROM transfer_chunks WHERE transfer_id = %s AND idx = %s", + (transfer_id, idx), + ).fetchone() + return dict(row) if row else None + + def is_chunk_received(self, transfer_id: str, idx: int) -> bool: + with self._conn() as conn: + row = conn.execute( + "SELECT 1 FROM transfer_chunks WHERE transfer_id = %s AND idx = %s AND received_at IS NOT NULL", + (transfer_id, idx), + ).fetchone() + return row is not None + + def mark_chunk_received(self, transfer_id: str, idx: int) -> None: + with self._conn() as conn: + conn.execute( + "UPDATE transfer_chunks SET received_at = now() " + "WHERE transfer_id = %s AND idx = %s", + (transfer_id, idx), + ) + + def get_received_chunks(self, transfer_id: str) -> set[int]: + with self._conn() as conn: + rows = conn.execute( + "SELECT idx FROM transfer_chunks WHERE transfer_id = %s AND received_at IS NOT NULL", + (transfer_id,), + ).fetchall() + return {r["idx"] for r in rows} + + # ---------- files ---------- + + def insert_file( + self, + transfer_id: str, + file_name: str, + path: str, + size: int, + sha256: str, + ) -> str: + file_id = uuid.uuid4().hex[:12] + with self._conn() as conn: + conn.execute( + "INSERT INTO files (file_id, transfer_id, file_name, path, size, sha256) " + "VALUES (%s, %s, %s, %s, %s, %s)", + (file_id, transfer_id, file_name, path, size, sha256), + ) + return file_id diff --git a/decrypt.py b/decrypt.py new file mode 100644 index 0000000..c91d062 --- /dev/null +++ b/decrypt.py @@ -0,0 +1,48 @@ +""" +解密引擎模块 (decrypt) +模块: 服务端 / 解密引擎 +输入: 密文主体 + 参数 +输出: 明文流 -> storage + +流式 AES-GCM (cobblestone) 解密, 块级认证即时检出篡改/密钥错误。 +""" + +from pathlib import Path + +from cryptography.cobblestone import Cobblestone256Decryptor +from cryptography.exceptions import InvalidTag + +from settings import CONTEXT +from keyring import Keyring, KeyringError + + +class DecryptError(RuntimeError): + """解密失败 (密钥错误/数据损坏)""" + + +class Decryptor: + def __init__(self, keyring: Keyring) -> None: + self.keyring = keyring + + def decrypt( + self, merged_path: Path, enc_params: dict, out_path: Path + ) -> Path: + """流式解密 merged -> out_path, 返回明文路径""" + key = self.keyring.get_key(enc_params["key_id"]) + context = enc_params.get("context", "").encode() or CONTEXT + try: + dec = Cobblestone256Decryptor(key, context) + with open(merged_path, "rb") as src, open(out_path, "wb") as dst: + while True: + chunk = src.read(1 << 16) + if not chunk: + break + dst.write(dec.update(chunk)) # 块级认证: 篡改在此抛 InvalidTag + dst.write(dec.finalize()) + except InvalidTag as e: + out_path.unlink(missing_ok=True) + raise DecryptError("解密验证失败: 密钥不匹配或密文被篡改") from e + except KeyringError as e: + out_path.unlink(missing_ok=True) + raise DecryptError(str(e)) from e + return out_path diff --git a/keyring.py b/keyring.py new file mode 100644 index 0000000..7e369ae --- /dev/null +++ b/keyring.py @@ -0,0 +1,57 @@ +""" +密钥管理模块 (keyring) +模块: 服务端 / 密钥管理 +输入: key_id +输出: 密钥材料 -> decrypt + +预共享密钥方案: 客户端与服务端使用同一 keyring (部署时复制), +key_id 标识密钥, 支持轮换 (新 key_id 用于新上传, 旧密钥保留仅解密)。 +""" + +import json +import os +from pathlib import Path + +from settings import KEYRING_PATH + + +class KeyringError(RuntimeError): + """密钥缺失或文件损坏""" + + +class Keyring: + def __init__(self, path: str | Path = KEYRING_PATH) -> None: + self.path = Path(path) + self._keys = self._load() + + def _load(self) -> dict[str, bytes]: + if not self.path.exists(): + return {} + try: + data = json.loads(self.path.read_text(encoding="utf-8")) + except json.JSONDecodeError as e: + raise KeyringError(f"密钥文件损坏: {self.path}") from e + return {kid: bytes.fromhex(hex_str) for kid, hex_str in data.items()} + + def _save(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text( + json.dumps({kid: key.hex() for kid, key in self._keys.items()}, indent=2), + encoding="utf-8", + ) + os.chmod(self.path, 0o600) + + def get_key(self, key_id: str) -> bytes: + """取密钥, 不存在抛 KeyringError (任务 failed 的根因之一)""" + try: + return self._keys[key_id] + except KeyError: + raise KeyringError(f"密钥ID '{key_id}' 不存在于服务端 keyring: {self.path}") from None + + def has_key(self, key_id: str) -> bool: + return key_id in self._keys + + def import_key(self, key_id: str, key: bytes) -> None: + """导入密钥 (部署时同步客户端 keyring)""" + self._keys[key_id] = key + self._save() diff --git a/main.py b/main.py new file mode 100644 index 0000000..fd7dfb6 --- /dev/null +++ b/main.py @@ -0,0 +1,14 @@ +"""服务端入口: uvicorn 启动 + +用法 (server/ 目录下): + .venv/bin/python main.py +或: + DATABASE_URL=... SZ_PORT=8000 .venv/bin/python main.py +""" + +import uvicorn + +from settings import HOST, PORT + +if __name__ == "__main__": + uvicorn.run("api:app", host=HOST, port=PORT, log_level="info") diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..02bae0e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,15 @@ +[project] +name = "7z-encrypt-server" +version = "0.1.0" +description = "文件安全传输系统 - 服务端 (接收/汇聚/解密/存储/入库)" +requires-python = ">=3.11" +dependencies = [ + "cryptography>=50.0", + "psycopg[binary]>=3.2", + "fastapi>=0.110", + "uvicorn>=0.30", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] diff --git a/receiver.py b/receiver.py new file mode 100644 index 0000000..b50f62e --- /dev/null +++ b/receiver.py @@ -0,0 +1,49 @@ +""" +分卷接收模块 (receiver) +模块: 服务端 / 分卷接收 +输入: 卷数据 (transfer_id, index) +输出: 落盘卷 + ok/409 + +逐卷校验 SHA-256 (对照 init 登记的规格); 重复卷直接 ok (幂等, 断点续传重发不炸)。 +""" + +import hashlib +import shutil +from pathlib import Path + +from task_manager import TaskManager + + +class ReceiverError(RuntimeError): + """卷校验失败 (哈希不一致/规格缺失)""" + + +class Receiver: + def __init__(self, tasks: TaskManager, tmp_root: str | Path) -> None: + self.tasks = tasks + self.tmp_root = Path(tmp_root) + + def receive(self, transfer_id: str, idx: int, data: bytes) -> None: + """接收并校验一卷。失败抛 ReceiverError (api 转 409)。""" + if self.tasks.is_chunk_received(transfer_id, idx): + return # 幂等: 重复卷直接 ok + spec = self.tasks.chunk_spec(transfer_id, idx) + if spec is None: + raise ReceiverError(f"任务 {transfer_id} 卷 {idx} 规格不存在") + + # 校验大小 + SHA-256 + if len(data) != spec["size"]: + raise ReceiverError(f"卷 {idx} 大小不符: 期望 {spec['size']}, 实际 {len(data)}") + sha256 = hashlib.sha256(data).hexdigest() + if sha256 != spec["sha256"]: + raise ReceiverError(f"卷 {idx} SHA-256 不一致 (数据损坏或串卷)") + + # 落盘临时目录 + chunk_dir = self.tmp_root / transfer_id + chunk_dir.mkdir(parents=True, exist_ok=True) + chunk_path = chunk_dir / f"chunk_{idx:04d}" + tmp_path = chunk_path.with_suffix(".tmp") + tmp_path.write_bytes(data) + shutil.move(str(tmp_path), str(chunk_path)) + + self.tasks.mark_chunk_received(transfer_id, idx) diff --git a/settings.py b/settings.py new file mode 100644 index 0000000..59e1985 --- /dev/null +++ b/settings.py @@ -0,0 +1,25 @@ +"""服务端配置 (环境变量驱动, 带本地默认值)""" +import os +from pathlib import Path + +# 应用绑定上下文 (必须与客户端一致) +CONTEXT = os.environ.get("SZ_CONTEXT", "7z-encrypt:v1").encode() + +# PostgreSQL 连接串 (本地默认走笔记本 PG socket) +DATABASE_URL = os.environ.get( + "DATABASE_URL", + "postgresql://lou@/7zencrypt?host=/home/lou/pgdata/socket", +) + +# 存储根目录 (明文落盘) +STORAGE_ROOT = Path(os.environ.get("SZ_STORAGE_ROOT", "server/data/storage")) + +# 临时目录 (分卷/合并/解密中间产物) +TMP_ROOT = Path(os.environ.get("SZ_TMP_ROOT", "server/data/tmp")) + +# 密钥库 (预共享: 部署时复制客户端 keyring 过来) +KEYRING_PATH = Path(os.environ.get("SZ_KEYRING", "server/data/keyring.json")) + +# API 服务 +HOST = os.environ.get("SZ_HOST", "0.0.0.0") +PORT = int(os.environ.get("SZ_PORT", "8000")) diff --git a/storage.py b/storage.py new file mode 100644 index 0000000..bfe0aaa --- /dev/null +++ b/storage.py @@ -0,0 +1,37 @@ +""" +存储层模块 (storage) +模块: 服务端 / 存储层 +输入: 明文流 + 元数据 +输出: 最终文件路径 + +按日期目录落盘, 同名冲突加后缀去重, 流式移动不整块进内存。 +""" + +import shutil +from datetime import datetime +from pathlib import Path + +from task_manager import TaskManager + + +class Storage: + def __init__(self, tasks: TaskManager, storage_root: str | Path) -> None: + self.tasks = tasks + self.storage_root = Path(storage_root) + + def store(self, plain_path: Path, file_name: str) -> Path: + """明文 -> 最终存储路径 (返回路径)""" + day_dir = self.storage_root / datetime.now().strftime("%Y-%m-%d") + 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(plain_path), str(target)) + return target diff --git a/task_manager.py b/task_manager.py new file mode 100644 index 0000000..c55226f --- /dev/null +++ b/task_manager.py @@ -0,0 +1,53 @@ +""" +任务管理模块 (task_manager) +模块: 服务端 / 任务管理 +输入: init 请求 / 卷到达事件 +输出: transfer 状态机 + +状态机: uploading -> assembling -> decrypting -> done / failed +transfer_id 生命周期贯穿接收->入库; 孤儿任务由清理逻辑回收 (见 server/data/tmp)。 +""" + +import uuid +from typing import Any + +from db import ServerDB + +STATUS_UPLOADING = "uploading" +STATUS_ASSEMBLING = "assembling" +STATUS_DECRYPTING = "decrypting" +STATUS_DONE = "done" +STATUS_FAILED = "failed" + + +class TaskManager: + """transfer 生命周期管理 (基于 ServerDB)""" + + def __init__(self, db: ServerDB) -> None: + self.db = db + + def create(self, init_json: dict[str, Any]) -> str: + """POST /init: 建任务, 返回 transfer_id""" + transfer_id = uuid.uuid4().hex[:12] + self.db.create_transfer(transfer_id, init_json) + return transfer_id + + def get(self, transfer_id: str) -> dict[str, Any] | None: + return self.db.get_transfer(transfer_id) + + def chunk_spec(self, transfer_id: str, idx: int) -> dict[str, Any] | None: + return self.db.get_chunk_spec(transfer_id, idx) + + def is_chunk_received(self, transfer_id: str, idx: int) -> bool: + return self.db.is_chunk_received(transfer_id, idx) + + def mark_chunk_received(self, transfer_id: str, idx: int) -> None: + """事件: 卷到达 (校验通过后)""" + self.db.mark_chunk_received(transfer_id, idx) + + def received(self, transfer_id: str) -> set[int]: + """已收卷集合 (GET /chunks 依据)""" + return self.db.get_received_chunks(transfer_id) + + def set_status(self, transfer_id: str, status: str) -> None: + self.db.set_status(transfer_id, status) diff --git a/unpacker.py b/unpacker.py new file mode 100644 index 0000000..51d9c02 --- /dev/null +++ b/unpacker.py @@ -0,0 +1,36 @@ +""" +解包器模块 (unpacker) +模块: 服务端 / 解包器 +输入: 完整密文文件 +输出: 密文主体 + 加密参数 + +cobblestone (C2SP chunked-encryption) 密文流自带头部+分块结构, +解包退化为前置结构检查: 非空 + 头部长度合法。真正的完整性由 decrypt 块级认证保证。 +""" + +from pathlib import Path + +from task_manager import TaskManager + +# cobblestone 头部最小长度 (magic + 版本 + 随机 nonce 等) +MIN_HEADER = 32 + + +class UnpackError(RuntimeError): + """密文包结构异常""" + + +class Unpacker: + def __init__(self, tasks: TaskManager) -> None: + self.tasks = tasks + + def unpack(self, transfer_id: str, merged_path: Path) -> tuple[Path, dict]: + """验证密文包结构, 返回 (密文路径, 加密参数)""" + transfer = self.tasks.get(transfer_id) + if transfer is None: + raise UnpackError(f"任务不存在: {transfer_id}") + size = merged_path.stat().st_size + if size < MIN_HEADER: + raise UnpackError(f"密文包过短 ({size}B < {MIN_HEADER}B), 结构损坏") + enc_params = dict(transfer["enc_params"]) + return merged_path, enc_params