# ============================================================ # WORM 归档模拟 + 完整性验证黄金测试（矩阵 104/112/116 的 WORM 剩余项） # 覆盖：归档生成 manifest（sha256 + merkle root + 归档时间 + 覆盖范围）、 # 归档后写拒绝（只读位 + manifest 在册）、同月二次归档不改写旧件、 # 篡改检出（tampered expected/actual）、缺失/无清单显式报告、 # merkle 复算校验、verify 汇总结构、网关端点接线（P0 verify / P1 run）。 # 说明：物理 WORM 介质（光盘/WORM 盘）为外部验收范围，本测试验证逻辑 WORM 语义。 # ============================================================ from __future__ import annotations import hashlib import json import os import stat from pathlib import Path import pytest from fastapi.testclient import TestClient from server.agent_core.audit_ledger import WORM_MANIFEST, WORM_NOTE, AnchorLedger from server.agent_core.audit_merkle import merkle_root def _event(i: int) -> dict: return {"id": i, "at": "2026-08-01T00:00:00+08:00", "actor": "tester", "category": "TEST", "action": f"test.{i}", "target": {"type": "TEST"}, "power": "P0", "rationale": {}, "prevHash": "GENESIS", "hash": "0" * 64} def _seed_ledger(ledger: AnchorLedger, n: int = 5) -> None: evs: list[dict] = [] for i in range(n): evs.append(_event(i + 1)) ledger.append(evs, reason="test") def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _manifest(ledger: AnchorLedger) -> dict: return json.loads((ledger.archive_dir / WORM_MANIFEST).read_text(encoding="utf-8")) def test_archive_run_generates_worm_manifest(tmp_path: Path): """归档生成 WORM 清单：每文件 sha256 + merkleRoot + 归档时间 + 覆盖范围 + 只读位。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger) r = ledger.archive_old(keep_last=2) assert r["archived"] == 3 assert r["worm"]["sealed"] is True assert r["worm"]["mode"] == "worm-simulated" manifest = _manifest(ledger) assert manifest["mode"] == "worm-simulated" assert WORM_NOTE in manifest["note"] # 文档化：物理介质为外部验收 fname = Path(r["archive"]).name entry = manifest["files"][fname] assert entry["sha256"] == _sha256(Path(r["archive"])) assert entry["merkleRoot"] # merkle root 摘要 assert entry["archivedAt"] # 归档时间 assert entry["coverage"]["records"] == 3 # 覆盖范围 assert entry["coverage"]["firstEventId"] and entry["coverage"]["lastEventId"] assert entry["readonly"] is True # WORM 只读位 def test_archive_file_write_refused_after_seal(tmp_path: Path): """归档后常规写路径拒绝改写：文件只读（直接写抛错）+ manifest 在册拒绝。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger) r = ledger.archive_old(keep_last=2) archive = Path(r["archive"]) # 常规文件写路径被只读位拒绝（Windows 抛 PermissionError） with pytest.raises((PermissionError, OSError)): archive.write_text(archive.read_text(encoding="utf-8") + "tamper\n", encoding="utf-8") # 账本写路径对已密封归档文件显式拒绝（manifest 在册 → ValueError） with pytest.raises(ValueError): ledger._guard_archive_sealed(archive) def test_second_archive_same_month_never_rewrites_sealed(tmp_path: Path): """同月二次归档不改写已密封文件：改用 ..jsonl，旧件 sha256 不变。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger, n=5) r1 = ledger.archive_old(keep_last=2) first = Path(r1["archive"]) first_sha = _sha256(first) # 追加新锚定记录后再次归档（同月）→ 新文件而非覆写旧件 _seed_ledger(ledger, n=3) r2 = ledger.archive_old(keep_last=2) second = Path(r2["archive"]) assert second.name != first.name assert second.name == f"{first.stem}.2.jsonl" or second.name.endswith(".2.jsonl") assert _sha256(first) == first_sha # 旧件字节级未变 manifest = _manifest(ledger) assert first.name in manifest["files"] and second.name in manifest["files"] def test_verify_archive_detects_tampering(tmp_path: Path): """篡改检出：改归档文件内容 → tampered {file, expected, actual}，ok=False。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger) r = ledger.archive_old(keep_last=2) archive = Path(r["archive"]) assert ledger.verify_archive()["ok"] is True os.chmod(archive, stat.S_IWRITE) # 模拟攻击者清除只读位 archive.write_text( archive.read_text(encoding="utf-8") + json.dumps({"eventId": "EVIL", "root": "0" * 64, "count": 1, "at": "2026-08-01 00:00", "reason": "tamper"}) + "\n", encoding="utf-8") result = ledger.verify_archive() assert result["ok"] is False assert result["summary"]["tampered"] == 1 item = result["tampered"][0] assert item["file"] == archive.name assert item["expected"] == r["worm"]["entry"]["sha256"] assert item["actual"] == _sha256(archive) assert item["actual"] != item["expected"] def test_verify_archive_reports_missing_and_no_manifest(tmp_path: Path): """缺失文件显式报告；无归档（无清单）给出确定性结构且 ok=True。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger) r = ledger.archive_old(keep_last=2) archive = Path(r["archive"]) os.chmod(archive, stat.S_IWRITE) # 清除只读位以便删除 archive.unlink() result = ledger.verify_archive() assert result["ok"] is False assert result["summary"]["missing"] == 1 assert result["missing"][0]["file"] == archive.name assert result["missing"][0]["expected"] # 清单中的期望 sha256 fresh = AnchorLedger("tenant-fresh", "fresh", ledger_dir=str(tmp_path / "other")) empty = fresh.verify_archive() assert empty["ok"] is True # 无归档 = 无篡改 assert empty["summary"]["noManifest"] is True assert empty["manifest"] is None def test_verify_archive_merkle_recompute_and_summary(tmp_path: Path): """merkle 复算一致 + verify 汇总字段齐全。""" ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path)) _seed_ledger(ledger) r = ledger.archive_old(keep_last=2) archive = Path(r["archive"]) records = [json.loads(ln) for ln in archive.read_text(encoding="utf-8").splitlines() if ln.strip()] manifest = _manifest(ledger) assert manifest["files"][archive.name]["merkleRoot"] == merkle_root(records) result = ledger.verify_archive() assert result["ok"] is True assert result["mode"] == "worm-simulated" s = result["summary"] for key in ("files", "verified", "tampered", "missing", "merkleMismatch", "unsealed", "noManifest"): assert key in s assert s["files"] == 1 and s["verified"] == 1 assert s["tampered"] == 0 and s["missing"] == 0 and s["merkleMismatch"] == 0 assert result["tampered"] == [] and result["missing"] == [] class _Store: def __init__(self): self.tenant_uuid = "platform" self.world_key = "default" self.data = {"auditEvents": [ {"id": 1, "at": "2026-08-01T00:00:00+08:00", "ts": "2026-08-01T00:00:00+08:00", "actor": "tester", "category": "TEST", "action": "test.one", "target": {"type": "TEST"}, "power": "P0", "rationale": {}, "prevHash": "GENESIS", "hash": "0" * 64}, ]} def next_id(self, _kind: str) -> int: return 2 def save(self) -> None: pass @pytest.fixture def client(monkeypatch, tmp_path): import server.gateway.app as gateway_module from tests.auth_provider import install_test_auth install_test_auth(monkeypatch, "tenant-worm-api") store = _Store() monkeypatch.setattr(gateway_module, "get_store", lambda: store) c = TestClient(gateway_module.create_app()) login = c.post("/api/auth/login", json={"username": "planner", "password": "test"}) assert login.status_code == 200, login.text return c, store def test_gov_archive_endpoints_roundtrip(client, tmp_path): """网关接线：/api/gov/archive/run 触发归档 + 生成 manifest；/verify 只读返回汇总明细。""" c, store = client os.environ["APS_AUDIT_LEDGER_DIR"] = str(tmp_path / "ledger") try: v0 = c.get("/api/gov/archive/verify") assert v0.status_code == 200 assert v0.json()["mode"] == "worm-simulated" assert v0.json()["summary"]["noManifest"] is True for _ in range(3): # 锚定 3 条账本记录 assert c.post("/api/gov/audit/anchor").status_code == 200 run = c.post("/api/gov/archive/run?keep_last=1") assert run.status_code == 200, run.text body = run.json() assert body["archived"] == 2 assert body["worm"]["sealed"] is True v = c.get("/api/gov/archive/verify") assert v.status_code == 200 data = v.json() assert data["ok"] is True assert data["summary"]["verified"] == 1 assert data["tampered"] == [] and data["missing"] == [] assert data["summary"]["noManifest"] is False finally: os.environ.pop("APS_AUDIT_LEDGER_DIR", None)