metadata 元数据生成模块: init json 组装+校验, 加密->分卷->init 全管线

.gitignore

@@ -14,3 +14,7 @@ kali-2026.zip
 
 # 分卷清单 (生成物)
 *_manifest.json
+
+# 分卷产物 / init json (生成物)
+*.part*
+*.init.json

metadata.py

@@ -0,0 +1,187 @@
+"""
+元数据生成模块 (metadata)
+模块：客户端 / 元数据生成
+输入：文件信息 + 加密参数 + 卷清单
+输出：init json (file_name/file_size/chunk_count/total_sha256/enc/chunks)
+
+init json 是服务端建任务的唯一依据 (POST /init 请求体)，字段缺一不可，
+组装后先自校验再输出。chunks 剥离本地 filename (服务端不关心客户端路径)。
+
+用法:
+    cd /home/lou/文档/7z-encrypt
+    .venv/bin/python metadata.py [文件] [--chunk-size MB] [--key-id 密钥ID]
+"""
+
+import argparse
+import hashlib
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+from crypto import CryptoEngine, KeyNotFoundError
+from splitter import split_stream
+
+SHA256_HEX_LEN = 64
+DEFAULT_CHUNK_MB = 10
+
+
+class MetadataError(ValueError):
+    """init json 字段缺失或非法"""
+
+
+def _require_str(obj: dict, key: str) -> str:
+    v = obj.get(key)
+    if not isinstance(v, str) or not v:
+        raise MetadataError(f"字段 '{key}' 缺失或非空字符串")
+    return v
+
+
+def _require_pos_int(obj: dict, key: str) -> int:
+    v = obj.get(key)
+    if not isinstance(v, int) or isinstance(v, bool) or v <= 0:
+        raise MetadataError(f"字段 '{key}' 缺失或非正整数")
+    return v
+
+
+def _require_sha256(obj: dict, key: str) -> str:
+    v = _require_str(obj, key)
+    if len(v) != SHA256_HEX_LEN:
+        raise MetadataError(f"字段 '{key}' 长度不是 {SHA256_HEX_LEN} (SHA-256 hex)")
+    return v
+
+
+def build_init_json(
+    file_name: str,
+    file_size: int,
+    total_sha256: str,
+    enc: dict[str, str],
+    chunks: list[dict],
+) -> dict[str, Any]:
+    """组装并校验 init json
+
+    Args:
+        file_name: 原始文件名
+        file_size: 原始文件字节数
+        total_sha256: 密文整体 SHA-256 (服务端合并后验证)
+        enc: 加密参数 {alg, key_id, context} (来自 crypto.encrypt_stream)
+        chunks: 卷清单 (来自 splitter, 含 filename; 组装时剥离)
+
+    Returns:
+        dict: init json {file_name, file_size, chunk_count, total_sha256, enc, chunks}
+
+    Raises:
+        MetadataError: 任一必填字段缺失/非法/卷序号不连续
+    """
+    # enc 校验
+    if not isinstance(enc, dict) or not enc:
+        raise MetadataError("字段 'enc' 缺失或为空")
+    for k in ("alg", "key_id", "context"):
+        _require_str(enc, k)
+
+    # chunks 校验 (序号必须 1..N 连续)
+    if not isinstance(chunks, list) or not chunks:
+        raise MetadataError("字段 'chunks' 缺失或为空")
+    clean_chunks: list[dict[str, Any]] = []
+    for i, ch in enumerate(chunks, start=1):
+        if not isinstance(ch, dict):
+            raise MetadataError(f"chunks[{i}] 不是对象")
+        if ch.get("index") != i:
+            raise MetadataError(
+                f"chunks[{i}] 序号不连续 (期望 {i}, 实际 {ch.get('index')})"
+            )
+        clean_chunks.append({
+            "index": i,
+            "size": _require_pos_int(ch, "size"),
+            "sha256": _require_sha256(ch, "sha256"),
+        })
+
+    # 整体校验
+    if not isinstance(file_name, str) or not file_name:
+        raise MetadataError("file_name 缺失或为空")
+    if not isinstance(file_size, int) or isinstance(file_size, bool) or file_size <= 0:
+        raise MetadataError("file_size 缺失或非法")
+    if len(total_sha256) != SHA256_HEX_LEN:
+        raise MetadataError(f"total_sha256 长度不是 {SHA256_HEX_LEN} (SHA-256 hex)")
+
+    return {
+        "file_name": file_name,
+        "file_size": file_size,
+        "chunk_count": len(clean_chunks),
+        "total_sha256": total_sha256,
+        "enc": enc,
+        "chunks": clean_chunks,
+    }
+
+
+def sha256_of(path: str | Path) -> str:
+    """流式计算文件 SHA-256 (不整块进内存)"""
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        while True:
+            chunk = f.read(1 << 16)
+            if not chunk:
+                break
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="元数据生成: 加密 -> 分卷 -> 组装 init json")
+    ap.add_argument("file", nargs="?", default="index.mp4", help="要处理的文件 (默认 index.mp4)")
+    ap.add_argument("--chunk-size", type=int, default=DEFAULT_CHUNK_MB,
+                    help=f"每卷大小 MB (默认 {DEFAULT_CHUNK_MB})")
+    ap.add_argument("--key-id", default="default_key", help="密钥ID (默认 default_key)")
+    args = ap.parse_args()
+
+    src_path = Path(args.file).resolve()
+    if not src_path.exists():
+        print(f"[错误] 文件不存在: {src_path}")
+        return 1
+
+    chunk_size = args.chunk_size * 1024 * 1024
+    enc_path = src_path.with_name(src_path.name + ".enc")
+    manifest_path = src_path.with_name(src_path.name + "_manifest.json")
+    init_path = src_path.with_name(src_path.name + ".init.json")
+
+    print("=== 元数据生成 (加密 -> 分卷 -> init json) ===")
+    print(f"文件: {src_path.name} ({src_path.stat().st_size / 1048576:.1f} MB), 卷大小: {args.chunk_size} MB")
+
+    # 1. 加密 -> 加密参数
+    engine = CryptoEngine()
+    try:
+        engine.get_key(args.key_id)
+    except KeyNotFoundError:
+        engine.generate_key(args.key_id)
+        print(f"[密钥库] 已生成新密钥 -> {engine.keyring_path}")
+    with open(src_path, "rb") as src, open(enc_path, "wb") as dst:
+        enc_params = engine.encrypt_stream(src, dst, args.key_id)
+    print(f"[1/4] 加密完成 -> {enc_path.name} ({enc_path.stat().st_size / 1048576:.1f} MB)")
+
+    # 2. 分卷 -> 卷清单
+    with open(enc_path, "rb") as f:
+        manifest = split_stream(f, chunk_size, str(enc_path), total_size=enc_path.stat().st_size)
+    print(f"[2/4] 分卷完成: {len(manifest)} 卷, 清单 -> {manifest_path.name}")
+
+    # 3. 密文整体 SHA-256
+    total_sha256 = sha256_of(enc_path)
+    print(f"[3/4] 密文 SHA-256: {total_sha256}")
+
+    # 4. 组装 init json
+    init_json = build_init_json(
+        file_name=src_path.name,
+        file_size=src_path.stat().st_size,
+        total_sha256=total_sha256,
+        enc=enc_params,
+        chunks=manifest,
+    )
+    with open(init_path, "w", encoding="utf-8") as f:
+        json.dump(init_json, f, indent=2, ensure_ascii=False)
+    print(f"[4/4] init json -> {init_path.name}")
+    print("\n=== init json ===")
+    print(json.dumps(init_json, indent=2, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_metadata.py

@@ -0,0 +1,115 @@
+"""metadata 模块行为测试 (组装/校验/集成管线, 非 mock)
+
+运行: cd /home/lou/文档/7z-encrypt && .venv/bin/python -m unittest discover -s tests -v
+"""
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from metadata import MetadataError, build_init_json, sha256_of
+
+PROJ = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PY = os.path.join(PROJ, '.venv/bin/python')
+
+
+def valid_chunks(n=3, start=1):
+    return [
+        {"index": i, "size": 100, "sha256": "a" * 64, "filename": f"p{i}.part{i:04d}"}
+        for i in range(start, start + n)
+    ]
+
+
+class TestBuildInitJson(unittest.TestCase):
+    def setUp(self):
+        self.enc = {"alg": "cobblestone-aes256gcm", "key_id": "k", "context": "7z-encrypt:v1"}
+
+    def test_valid(self):
+        out = build_init_json("a.mp4", 300, "b" * 64, self.enc, valid_chunks())
+        self.assertEqual(out["file_name"], "a.mp4")
+        self.assertEqual(out["file_size"], 300)
+        self.assertEqual(out["chunk_count"], 3)
+        self.assertEqual(out["total_sha256"], "b" * 64)
+        self.assertEqual(out["enc"], self.enc)
+        # chunks 剥离 filename
+        self.assertEqual(out["chunks"], [
+            {"index": 1, "size": 100, "sha256": "a" * 64},
+            {"index": 2, "size": 100, "sha256": "a" * 64},
+            {"index": 3, "size": 100, "sha256": "a" * 64},
+        ])
+
+    def test_enc_missing(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "b" * 64, {}, valid_chunks())
+
+    def test_enc_key_missing(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "b" * 64, {"alg": "x"}, valid_chunks())
+
+    def test_chunks_empty(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "b" * 64, self.enc, [])
+
+    def test_chunks_index_gap(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "b" * 64, self.enc, valid_chunks(start=2))
+
+    def test_chunks_bad_sha256_len(self):
+        ch = valid_chunks()
+        ch[0]["sha256"] = "abc"
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "b" * 64, self.enc, ch)
+
+    def test_bad_total_sha256(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 300, "short", self.enc, valid_chunks())
+
+    def test_file_size_zero(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("a.mp4", 0, "b" * 64, self.enc, valid_chunks())
+
+    def test_empty_file_name(self):
+        with self.assertRaises(MetadataError):
+            build_init_json("", 300, "b" * 64, self.enc, valid_chunks())
+
+
+class TestIntegration(unittest.TestCase):
+    """集成: 小文件 加密 -> 分卷 -> init json, 全部真实跑"""
+
+    def test_pipeline(self):
+        with tempfile.TemporaryDirectory(prefix='hermes-test-') as td:
+            data = os.urandom(3 << 20)  # 3MB
+            src = os.path.join(td, 'sample.bin')
+            with open(src, 'wb') as f:
+                f.write(data)
+            r = subprocess.run(
+                [PY, os.path.join(PROJ, 'metadata.py'), src, '--chunk-size', '1'],
+                cwd=PROJ, capture_output=True, text=True, timeout=180,
+            )
+            self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+            init_path = src + '.init.json'
+            with open(init_path, encoding='utf-8') as f:
+                init = json.load(f)
+            self.assertEqual(init['file_name'], 'sample.bin')
+            self.assertEqual(init['file_size'], len(data))
+            enc_size = os.path.getsize(src + '.enc')
+            # 密文含 header+tag 开销, 卷数按密文实际大小算
+            import math
+            self.assertEqual(init['chunk_count'], math.ceil(enc_size / (1 << 20)))
+            self.assertEqual(len(init['chunks']), init['chunk_count'])
+            self.assertEqual(len(init['total_sha256']), 64)
+            # 密文总哈希 = init 里的 total_sha256
+            self.assertEqual(init['total_sha256'], sha256_of(src + '.enc'))
+            # 卷清单哈希与真实卷文件一致
+            for ch in init['chunks']:
+                part = os.path.join(td, os.path.basename(src) + '.enc.part%04d' % ch['index'])
+                self.assertEqual(ch['sha256'], sha256_of(part))
+
+
+if __name__ == '__main__':
+    unittest.main()