123 lines
5.3 KiB
Python
123 lines
5.3 KiB
Python
# 审计独立介质锚定黄金测试（plan.md §3.6 / 矩阵「全审计」）
|
||
# ============================================================
|
||
# 覆盖：根确定性、锚定往返、篡改检测（改事件/删账本行/改根）、租户/项目隔离。
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
from pathlib import Path
|
||
|
||
from server.agent_core.audit_ledger import AnchorLedger, audit_root, GENESIS_ROOT
|
||
|
||
|
||
def _events(n: int = 3) -> list[dict]:
|
||
out: list[dict] = []
|
||
prev = "GENESIS"
|
||
import hashlib
|
||
for i in range(n):
|
||
ev = {"id": i + 1, "action": f"action-{i}", "hash": prev}
|
||
prev = hashlib.sha256((prev + json.dumps(ev, sort_keys=True)).encode()).hexdigest()
|
||
ev["hash"] = prev
|
||
out.append(ev)
|
||
return out
|
||
|
||
|
||
def test_audit_root_is_deterministic_and_order_sensitive():
|
||
"""锚定根确定性：相同事件同根；顺序/数量变化则不同（root 聚合链哈希字段）。"""
|
||
events = _events()
|
||
assert audit_root(events) == audit_root(events)
|
||
assert audit_root([]) == GENESIS_ROOT
|
||
assert audit_root(events) != audit_root(events[:-1]) # 数量变化
|
||
assert audit_root(events) != audit_root(list(reversed(events))) # 顺序变化
|
||
# root 聚合 hash 字段：哈希序列变化（篡改 hash 本身）必然改变 root
|
||
tampered_hash = copy.deepcopy(events)
|
||
tampered_hash[1]["hash"] = "0" * 64
|
||
assert audit_root(events) != audit_root(tampered_hash)
|
||
|
||
|
||
def test_audit_root_plus_chain_catches_body_tampering():
|
||
"""事件体篡改由链校验捕捉，root 保持不变（root 信任链哈希，二者互补）。"""
|
||
from server.agent_core.registry import verify_audit_chain
|
||
events = _events()
|
||
tampered = copy.deepcopy(events)
|
||
tampered[1]["action"] = "hacked" # 改事件体但不动 hash
|
||
assert audit_root(events) == audit_root(tampered) # root 不变
|
||
assert verify_audit_chain(tampered)["ok"] is False # 链校验失败
|
||
|
||
|
||
def test_ledger_append_and_status_match(tmp_path: Path):
|
||
"""锚定往返：append 后 status 反映 ok=true 与最新根。"""
|
||
events = _events()
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
assert ledger.status(events)["anchored"] is False # 未锚定
|
||
record = ledger.append(events, reason="test")
|
||
assert record["count"] == len(events)
|
||
status = ledger.status(events)
|
||
assert status["anchored"] is True
|
||
assert status["ok"] is True
|
||
assert status["currentRoot"] == record["root"]
|
||
|
||
|
||
def test_ledger_detects_event_tampering(tmp_path: Path):
|
||
"""篡改检测：锚定后改 hash 序列 → ok=false（根不匹配）。"""
|
||
events = _events()
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
ledger.append(events)
|
||
tampered = copy.deepcopy(events)
|
||
tampered[2]["hash"] = "f" * 64 # 哈希被改 → root 变化
|
||
status = ledger.status(tampered)
|
||
assert status["ok"] is False
|
||
assert status["reason"] == "mismatch"
|
||
|
||
|
||
def test_ledger_detects_deleted_ledger_line(tmp_path: Path):
|
||
"""不可改删：删除账本行后 status 退化（记录数减少/无记录）。"""
|
||
events = _events()
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
ledger.append(events)
|
||
ledger.append(events, reason="second")
|
||
assert ledger.status(events)["records"] == 2
|
||
# 删除最后一行（模拟介质被删改）
|
||
lines = ledger.path.read_text(encoding="utf-8").splitlines()
|
||
ledger.path.write_text("\n".join(lines[:-1]) + "\n", encoding="utf-8")
|
||
status = ledger.status(events)
|
||
assert status["records"] == 1
|
||
# 仍锚定但可能 mismatch（若删除的是最新匹配行）
|
||
assert status["anchored"] is True
|
||
|
||
|
||
def test_ledger_isolation_per_tenant_world(tmp_path: Path):
|
||
"""租户/项目隔离：不同 scope 各自独立账本文件。"""
|
||
events = _events()
|
||
a = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
b = AnchorLedger("tenant-b", "world-2", ledger_dir=str(tmp_path))
|
||
a.append(events)
|
||
assert b.status(events)["anchored"] is False
|
||
assert a.path != b.path
|
||
assert a.path.exists() and not b.path.exists()
|
||
|
||
|
||
def test_ledger_archive_generates_worm_manifest(tmp_path: Path):
|
||
"""WORM 模拟：archive_old 归档超窗记录并生成 .worm-manifest.json（sha256 + merkle）。"""
|
||
import json as _json
|
||
from server.agent_core.audit_ledger import WORM_MANIFEST
|
||
from server.agent_core.audit_merkle import merkle_root
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
evs = []
|
||
for i in range(5):
|
||
evs.append({"id": i + 1, "action": f"a{i}", "hash": "0" * 64})
|
||
ledger.append(evs, reason="test")
|
||
r = ledger.archive_old(keep_last=2)
|
||
assert r["archived"] == 3 and r["kept"] == 2
|
||
assert r["worm"]["sealed"] is True
|
||
archive = Path(r["archive"])
|
||
manifest_path = archive.parent / WORM_MANIFEST
|
||
assert manifest_path.exists()
|
||
manifest = _json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
entry = manifest["files"][archive.name]
|
||
assert entry["sha256"] and entry["merkleRoot"]
|
||
records = [_json.loads(ln) for ln in archive.read_text(encoding="utf-8").splitlines()
|
||
if ln.strip()]
|
||
assert entry["merkleRoot"] == merkle_root(records)
|
||
assert ledger.verify_archive()["ok"] is True
|