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
|