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
|