lou
394 lines
17 KiB
Python
394 lines
17 KiB
Python
"""服务端集成测试: 真实 FastAPI + SQLite + 真实密码学管线 (零知识架构)
|
|
|
|
覆盖: 全链路 / 坏卷 409 / 缺卷 409 / 幂等 / 404 / 认证 401 / ls / 下载密文
|
|
"""
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
# 测试用配置 (先于 api 导入)
|
|
os.environ["SZ_DB_PATH"] = "/tmp/sz-test.db"
|
|
os.environ["SZ_TMP_ROOT"] = "/tmp/sz-test-tmp"
|
|
os.environ["SZ_STORAGE_ROOT"] = "/tmp/sz-test-storage"
|
|
os.environ["SZ_TOKEN"] = "test-token"
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from db import ServerDB # noqa: E402
|
|
|
|
# 客户端模块 (7z-encrypt 项目)
|
|
CLIENT_DIR = "/home/lou/文档/7z-encrypt"
|
|
sys.path.insert(0, CLIENT_DIR)
|
|
from crypto import CryptoEngine # type: ignore[import-not-found] # noqa: E402
|
|
from metadata import build_init_json # type: ignore[import-not-found] # noqa: E402
|
|
from splitter import split_stream # type: ignore[import-not-found] # noqa: E402
|
|
|
|
TEST_KEY_ID = "test_key"
|
|
AUTH = {"Authorization": "Bearer test-token"}
|
|
|
|
|
|
def _prepare(keyring_path: str, data: bytes, chunk_size: int = 1 << 20):
|
|
"""客户端管线: 加密 -> 分卷 -> init json, 返回 (init_json, chunk_files, enc_bytes)"""
|
|
engine = CryptoEngine(keyring_path)
|
|
enc = io.BytesIO()
|
|
params = engine.encrypt_stream(io.BytesIO(data), enc, TEST_KEY_ID)
|
|
enc_bytes = enc.getvalue()
|
|
total_sha256 = hashlib.sha256(enc_bytes).hexdigest()
|
|
|
|
tmpdir = tempfile.mkdtemp(prefix="sz-chunks-")
|
|
manifest = split_stream(io.BytesIO(enc_bytes), chunk_size, os.path.join(tmpdir, "c"))
|
|
chunk_files = {ch["index"]: ch["filename"] for ch in manifest}
|
|
init_json = build_init_json("test.bin", len(data), total_sha256, params, manifest)
|
|
return init_json, chunk_files, enc_bytes
|
|
|
|
|
|
class ServerPipelineTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# 清空测试库 (SQLite: 删文件重建)
|
|
for f in ("/tmp/sz-test.db", "/tmp/sz-test.db-wal", "/tmp/sz-test.db-shm"):
|
|
os.path.exists(f) and os.unlink(f)
|
|
# 清空运行时目录 (零知识: 无 keyring)
|
|
for d in ("/tmp/sz-test-tmp", "/tmp/sz-test-storage"):
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
# 客户端密钥 (仅客户端持有, 服务端零知识)
|
|
client_kr = os.path.join(tempfile.mkdtemp(prefix="sz-kr-"), "keyring.json")
|
|
engine = CryptoEngine(client_kr)
|
|
engine.generate_key(TEST_KEY_ID)
|
|
cls.client_kr = client_kr
|
|
|
|
from api import app
|
|
cls.client = TestClient(app)
|
|
|
|
def _upload(self, init_json, chunk_files):
|
|
r = self.client.post("/api/transfer/init", json=init_json, headers=AUTH)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
tid = r.json()["transfer_id"]
|
|
for idx in sorted(chunk_files):
|
|
with open(chunk_files[idx], "rb") as f:
|
|
r = self.client.put(
|
|
f"/api/transfer/{tid}/chunk/{idx}", content=f.read(), headers=AUTH
|
|
)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
r = self.client.get(f"/api/transfer/{tid}/chunks", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(len(r.json()["received"]), len(chunk_files))
|
|
return tid
|
|
|
|
# ---------- 认证 ----------
|
|
|
|
def test_auth_required(self):
|
|
# 无 token / 错 token -> 401
|
|
self.assertEqual(self.client.get("/api/files").status_code, 401)
|
|
self.assertEqual(
|
|
self.client.post("/api/transfer/init", json={}).status_code, 401
|
|
)
|
|
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
|
|
init_json, chunk_files, enc_bytes = _prepare(self.client_kr, data)
|
|
tid = self._upload(init_json, chunk_files)
|
|
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(r.json()["status"], "done")
|
|
file_id = r.json()["file_id"]
|
|
self.assertTrue(file_id)
|
|
|
|
# 零合并: 存储的是卷目录 (chunk_XXXX 文件), 不是合并单文件, 没有明文
|
|
db = ServerDB()
|
|
with db._conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT path FROM files WHERE file_id = ?", (file_id,)
|
|
).fetchone()
|
|
self.assertIsNotNone(row)
|
|
chunk_dir = Path(row["path"])
|
|
self.assertTrue(chunk_dir.is_dir())
|
|
# 每用户独立目录: admin 上传 -> storage/admin/日期/transfer_id
|
|
self.assertIn("admin", chunk_dir.parts)
|
|
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.assertNotEqual(stored, data) # 服务端手里不是明文
|
|
|
|
# transfer 状态 done
|
|
with db._conn() as conn:
|
|
st = conn.execute(
|
|
"SELECT status FROM transfers WHERE transfer_id = ?", (tid,)
|
|
).fetchone()
|
|
self.assertEqual(st["status"], "done")
|
|
|
|
def test_init_sanitizes_file_name(self):
|
|
# 恶意文件名: 控制字符删除 + 长度截断 (路径分隔符保留, 由客户端 _safe_name 兜底)
|
|
init_json, chunk_files, _ = _prepare(
|
|
self.client_kr, os.urandom(64 << 10)
|
|
)
|
|
init_json["file_name"] = "../../x/..\\y\x00z" * 50
|
|
tid = self._upload(init_json, chunk_files)
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
file_id = r.json()["file_id"]
|
|
db = ServerDB()
|
|
with db._conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT file_name FROM files WHERE file_id = ?", (file_id,)
|
|
).fetchone()
|
|
name = row["file_name"]
|
|
self.assertNotIn("\x00", name) # 控制字符被删
|
|
self.assertLessEqual(len(name), 255)
|
|
self.assertIn("..", name) # 路径段保留 (加密名含 - _ 不受影响, 明文名由下载端消毒)
|
|
|
|
def test_init_oversized_json_rejected(self):
|
|
# init json 超大 -> 413
|
|
import json as _json
|
|
data = os.urandom(8 << 10)
|
|
init_json, _, _ = _prepare(self.client_kr, data)
|
|
init_json["chunks"] = [
|
|
{"index": i, "size": 1, "sha256": "0" * 64} for i in range(10001)
|
|
]
|
|
r = self.client.post(
|
|
"/api/transfer/init", headers=AUTH, content=_json.dumps(init_json)
|
|
)
|
|
self.assertIn(r.status_code, (400, 413))
|
|
|
|
def test_bad_chunk_rejected(self):
|
|
data = os.urandom(1 << 20)
|
|
init_json, chunk_files, _ = _prepare(self.client_kr, data, chunk_size=512 * 1024)
|
|
r = self.client.post("/api/transfer/init", json=init_json, headers=AUTH)
|
|
tid = r.json()["transfer_id"]
|
|
# 篡改第一卷
|
|
bad = bytearray(open(chunk_files[1], "rb").read())
|
|
bad[10] ^= 0xFF
|
|
r = self.client.put(f"/api/transfer/{tid}/chunk/1", content=bytes(bad), headers=AUTH)
|
|
self.assertEqual(r.status_code, 409)
|
|
|
|
def test_incomplete_complete_409(self):
|
|
data = os.urandom(1 << 20)
|
|
init_json, chunk_files, _ = _prepare(self.client_kr, data, chunk_size=512 * 1024)
|
|
r = self.client.post("/api/transfer/init", json=init_json, headers=AUTH)
|
|
tid = r.json()["transfer_id"]
|
|
# 只传 1 卷就 complete
|
|
with open(chunk_files[1], "rb") as f:
|
|
self.client.put(
|
|
f"/api/transfer/{tid}/chunk/1", content=f.read(), headers=AUTH
|
|
)
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=AUTH)
|
|
self.assertEqual(r.status_code, 409)
|
|
|
|
def test_duplicate_chunk_idempotent(self):
|
|
data = os.urandom(512 * 1024)
|
|
init_json, chunk_files, _ = _prepare(self.client_kr, data, chunk_size=256 * 1024)
|
|
r = self.client.post("/api/transfer/init", json=init_json, headers=AUTH)
|
|
tid = r.json()["transfer_id"]
|
|
with open(chunk_files[1], "rb") as f:
|
|
payload = f.read()
|
|
self.assertEqual(
|
|
self.client.put(
|
|
f"/api/transfer/{tid}/chunk/1", content=payload, headers=AUTH
|
|
).status_code,
|
|
200,
|
|
)
|
|
# 重复传同一卷 (断点续传重发) -> 幂等 ok
|
|
self.assertEqual(
|
|
self.client.put(
|
|
f"/api/transfer/{tid}/chunk/1", content=payload, headers=AUTH
|
|
).status_code,
|
|
200,
|
|
)
|
|
|
|
def test_unknown_transfer_404(self):
|
|
self.assertEqual(
|
|
self.client.get("/api/transfer/nope/chunks", headers=AUTH).status_code, 404
|
|
)
|
|
self.assertEqual(
|
|
self.client.put(
|
|
"/api/transfer/nope/chunk/1", content=b"x", headers=AUTH
|
|
).status_code,
|
|
404,
|
|
)
|
|
|
|
def test_list_and_download(self):
|
|
data = os.urandom(512 * 1024)
|
|
init_json, chunk_files, enc_bytes = _prepare(
|
|
self.client_kr, data, chunk_size=256 * 1024
|
|
)
|
|
tid = self._upload(init_json, chunk_files)
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
file_id = r.json()["file_id"]
|
|
|
|
# ls 列表包含刚上传的文件
|
|
r = self.client.get("/api/files", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
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}/chunk/1", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
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/chunk/1", headers=AUTH).status_code, 404
|
|
)
|
|
|
|
# complete 幂等: 重复 complete 返回同一个 file_id, 不新增入库记录
|
|
before = len(self.client.get("/api/files", headers=AUTH).json()["files"])
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()["file_id"], file_id)
|
|
self.assertEqual(
|
|
len(self.client.get("/api/files", headers=AUTH).json()["files"]), before
|
|
)
|
|
|
|
# 配额: 返回结构 + 已用 = files size 总和
|
|
r = self.client.get("/api/quota", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
q = r.json()
|
|
self.assertIn("used_bytes", q)
|
|
self.assertIn("quota_bytes", q)
|
|
self.assertIn("remain_bytes", q)
|
|
self.assertEqual(q["used_bytes"], q["quota_bytes"] - q["remain_bytes"])
|
|
self.assertGreater(q["quota_bytes"], 0)
|
|
|
|
# del: 删除后列表减少, 再删 404
|
|
r = self.client.delete(f"/api/files/{file_id}", headers=AUTH)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertTrue(r.json()["ok"])
|
|
self.assertEqual(
|
|
len(self.client.get("/api/files", headers=AUTH).json()["files"]), before - 1
|
|
)
|
|
self.assertEqual(
|
|
self.client.delete(f"/api/files/{file_id}", headers=AUTH).status_code, 404
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class AccountTest(unittest.TestCase):
|
|
"""账号体系: 注册 / 登录 / 注销 / 文件归属隔离"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# 清库重建 (账号测试需要干净的用户表, 防跨运行残留)
|
|
for f in ("/tmp/sz-test.db", "/tmp/sz-test.db-wal", "/tmp/sz-test.db-shm"):
|
|
os.path.exists(f) and os.unlink(f)
|
|
from db import ServerDB
|
|
ServerDB() # 重建 schema (api 全局 db 实例后续连接即见新表)
|
|
from api import app
|
|
cls.client = TestClient(app)
|
|
# 客户端密钥 (测试用, 与 ServerPipelineTest 同款)
|
|
client_kr = os.path.join(tempfile.mkdtemp(prefix="sz-acc-kr-"), "keyring.json")
|
|
engine = CryptoEngine(client_kr)
|
|
engine.generate_key(TEST_KEY_ID)
|
|
cls.client_kr = client_kr
|
|
|
|
def _register(self, name: str, pw: str = "pass123456"):
|
|
return self.client.post(
|
|
"/api/auth/register", json={"username": name, "password": pw}
|
|
)
|
|
|
|
def test_register_login_logout(self):
|
|
# 注册 -> 自动登录返回 token
|
|
r = self._register("alice")
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
token = r.json()["token"]
|
|
self.assertTrue(token)
|
|
auth = {"Authorization": f"Bearer {token}"}
|
|
# token 可用
|
|
self.assertEqual(self.client.get("/api/files", headers=auth).status_code, 200)
|
|
# 重名注册 -> 409
|
|
self.assertEqual(self._register("alice").status_code, 409)
|
|
# 注销 -> token 失效 -> 401
|
|
self.assertEqual(self.client.post("/api/auth/logout", headers=auth).status_code, 200)
|
|
self.assertEqual(self.client.get("/api/files", headers=auth).status_code, 401)
|
|
# 重新登录 -> 新 token 可用
|
|
r = self.client.post(
|
|
"/api/auth/login", json={"username": "alice", "password": "pass123456"}
|
|
)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
auth2 = {"Authorization": f"Bearer {r.json()['token']}"}
|
|
self.assertEqual(self.client.get("/api/files", headers=auth2).status_code, 200)
|
|
|
|
def test_register_validation(self):
|
|
# 用户名/密码规则
|
|
self.assertEqual(self._register("ab", "pass123456").status_code, 400) # 太短
|
|
self.assertEqual(self._register("a b c", "pass123456").status_code, 400) # 非字母数字
|
|
self.assertEqual(self._register("bob", "12345").status_code, 400) # 密码太短
|
|
# 错误密码登录 -> 401
|
|
r = self.client.post(
|
|
"/api/auth/login", json={"username": "bob", "password": "wrongpass"}
|
|
)
|
|
self.assertEqual(r.status_code, 401)
|
|
|
|
def test_user_file_isolation(self):
|
|
# alice 上传文件, bob 看不到也不能下载/删除
|
|
init_json, chunk_files, _ = _prepare(self.client_kr, os.urandom(256 << 10))
|
|
ra = self.client.post(
|
|
"/api/auth/register", json={"username": "u_isola", "password": "pass123456"}
|
|
)
|
|
auth_a = {"Authorization": f"Bearer {ra.json()['token']}"}
|
|
rb = self.client.post(
|
|
"/api/auth/register", json={"username": "u_isolb", "password": "pass123456"}
|
|
)
|
|
auth_b = {"Authorization": f"Bearer {rb.json()['token']}"}
|
|
|
|
r = self.client.post("/api/transfer/init", json=init_json, headers=auth_a)
|
|
tid = r.json()["transfer_id"]
|
|
for idx in sorted(chunk_files):
|
|
with open(chunk_files[idx], "rb") as f:
|
|
self.client.put(
|
|
f"/api/transfer/{tid}/chunk/{idx}", content=f.read(), headers=auth_a
|
|
)
|
|
r = self.client.post(f"/api/transfer/{tid}/complete", headers=auth_a)
|
|
file_id = r.json()["file_id"]
|
|
|
|
# bob 的 ls 看不到 alice 的文件
|
|
r = self.client.get("/api/files", headers=auth_b)
|
|
self.assertTrue(all(f["file_id"] != file_id for f in r.json()["files"]))
|
|
# bob 下载 alice 的文件 -> 403
|
|
r = self.client.get(f"/api/files/{file_id}/chunk/1", headers=auth_b)
|
|
self.assertEqual(r.status_code, 403)
|
|
# bob 删除 alice 的文件 -> 403
|
|
r = self.client.delete(f"/api/files/{file_id}", headers=auth_b)
|
|
self.assertEqual(r.status_code, 403)
|
|
# alice 自己可以下载/删除
|
|
r = self.client.get(f"/api/files/{file_id}/chunk/1", headers=auth_a)
|
|
self.assertEqual(r.status_code, 200)
|
|
# admin (SZ_TOKEN) 可见全部
|
|
r = self.client.get("/api/files", headers=AUTH)
|
|
self.assertTrue(any(f["file_id"] == file_id for f in r.json()["files"]))
|