117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
|
# 审计事件独立介质镜像黄金测试（plan.md §3.6 / 矩阵「审计 append-only」）
|
|||
|
|
# ============================================================
|
|||
|
|
# 覆盖：镜像追加/读取、改 world 不影响镜像、租户隔离、启用/禁用开关、
|
|||
|
|
# 链校验仍可作用于镜像事件、gov_audit 镜像优先。
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import copy
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
from server.agent_core import audit
|
|||
|
|
from server.agent_core.audit_mirror import AuditMirror, mirror_enabled
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _event(i: int = 1) -> dict:
|
|||
|
|
return {"id": i, "at": "2026-08-01T00:00:00+08:00", "ts": "2026-08-01T00:00:00+08:00",
|
|||
|
|
"actor": "tester", "category": "TEST", "action": f"test.{i}",
|
|||
|
|
"target": {"type": "TEST"}, "power": "P0", "rationale": {},
|
|||
|
|
"evidenceRefs": [], "beforeSnapshot": None, "result": "SUCCESS",
|
|||
|
|
"prevHash": "GENESIS", "hash": "0" * 64}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_mirror_append_and_read(tmp_path: Path):
|
|||
|
|
"""镜像追加/读取：追加事件可读回，计数正确。"""
|
|||
|
|
m = AuditMirror("tenant-a", "world-1", mirror_dir=str(tmp_path))
|
|||
|
|
assert m.read_events() == []
|
|||
|
|
m.append_event(_event(1))
|
|||
|
|
m.append_event(_event(2))
|
|||
|
|
events = m.read_events()
|
|||
|
|
assert len(events) == 2
|
|||
|
|
assert [e["id"] for e in events] == [1, 2]
|
|||
|
|
assert m.count() == 2
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_mirror_isolated_per_tenant_world(tmp_path: Path):
|
|||
|
|
"""租户/项目隔离：不同 scope 各自独立文件。"""
|
|||
|
|
a = AuditMirror("tenant-a", "world-1", mirror_dir=str(tmp_path))
|
|||
|
|
b = AuditMirror("tenant-b", "world-2", mirror_dir=str(tmp_path))
|
|||
|
|
a.append_event(_event(1))
|
|||
|
|
assert b.read_events() == []
|
|||
|
|
assert a.path != b.path
|
|||
|
|
assert a.path.exists() and not b.path.exists()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_mirror_unaffected_by_world_mutation(tmp_path: Path, monkeypatch):
|
|||
|
|
"""改 world 不影响镜像：镜像保留原始事件（独立介质证据）。"""
|
|||
|
|
m = AuditMirror("tenant-a", "world-1", mirror_dir=str(tmp_path))
|
|||
|
|
m.append_event(_event(1))
|
|||
|
|
# 模拟业务进程篡改 world 副本
|
|||
|
|
forged = copy.deepcopy(_event(1))
|
|||
|
|
forged["actor"] = "hacker"
|
|||
|
|
world = {"auditEvents": [forged]}
|
|||
|
|
del world # world 修改不影响镜像
|
|||
|
|
events = m.read_events()
|
|||
|
|
assert events[0]["actor"] == "tester" # 镜像保留原始值
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_mirror_enabled_switch(tmp_path: Path, monkeypatch):
|
|||
|
|
"""启用/禁用开关：APS_AUDIT_MIRROR=1 启用，缺省禁用。"""
|
|||
|
|
monkeypatch.delenv("APS_AUDIT_MIRROR", raising=False)
|
|||
|
|
monkeypatch.delenv("APS_AUDIT_MIRROR_DIR", raising=False)
|
|||
|
|
assert mirror_enabled() is False
|
|||
|
|
monkeypatch.setenv("APS_AUDIT_MIRROR", "1")
|
|||
|
|
assert mirror_enabled() is True
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_write_audit_mirrors_when_enabled(tmp_path: Path, monkeypatch):
|
|||
|
|
"""write_audit 启用镜像时把事件写入独立介质。"""
|
|||
|
|
monkeypatch.setenv("APS_AUDIT_MIRROR", "1")
|
|||
|
|
monkeypatch.setenv("APS_AUDIT_MIRROR_DIR", str(tmp_path / "mirror"))
|
|||
|
|
monkeypatch.setattr("server.auth.context.get_identity", lambda: _FakeIdentity())
|
|||
|
|
import server.state.projects as projects_module
|
|||
|
|
monkeypatch.setattr(projects_module, "get_project_store", lambda: _FakeProjectStore())
|
|||
|
|
|
|||
|
|
world: dict = {}
|
|||
|
|
audit.write_audit(world, lambda _k: 1, actor="tester", category="TEST",
|
|||
|
|
action="test.mirror", target={"type": "T"},
|
|||
|
|
power="P0", rationale={})
|
|||
|
|
assert world["auditEvents"][0]["id"] == 1
|
|||
|
|
m = AuditMirror("tenant-fake", "world-fake", mirror_dir=str(tmp_path / "mirror"))
|
|||
|
|
events = m.read_events()
|
|||
|
|
assert len(events) == 1
|
|||
|
|
assert events[0]["action"] == "test.mirror"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _FakeIdentity:
|
|||
|
|
tenant_uuid = "tenant-fake"
|
|||
|
|
user_id = None
|
|||
|
|
username = None
|
|||
|
|
fullname = None
|
|||
|
|
roles = ()
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _FakeProjectStore:
|
|||
|
|
def active_world_key(self) -> str:
|
|||
|
|
return "world-fake"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_mirror_events_pass_chain_verify(tmp_path: Path):
|
|||
|
|
"""链校验仍可作用于镜像事件（镜像保留完整哈希链）。"""
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
|
|||
|
|
from server.agent_core.registry import verify_audit_chain
|
|||
|
|
prev = "GENESIS"
|
|||
|
|
events = []
|
|||
|
|
m = AuditMirror("tenant-a", "world-1", mirror_dir=str(tmp_path))
|
|||
|
|
for i in range(3):
|
|||
|
|
ev = _event(i + 1)
|
|||
|
|
ev["prevHash"] = prev # 链式前驱（第二事件起指向前一事件 hash）
|
|||
|
|
body = {k: v for k, v in ev.items() if k != "hash"} # 与 verify_audit_chain 剥离方式一致
|
|||
|
|
payload = prev + json.dumps(body, ensure_ascii=False, sort_keys=True)
|
|||
|
|
ev["hash"] = hashlib.sha256(payload.encode()).hexdigest()
|
|||
|
|
prev = ev["hash"]
|
|||
|
|
events.append(ev)
|
|||
|
|
m.append_event(ev)
|
|||
|
|
assert verify_audit_chain(m.read_events())["ok"] is True
|