116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
# 审计合规导出与保留脱敏黄金测试（plan.md §3.6 / 矩阵「全审计」）
|
||
# ============================================================
|
||
# 覆盖：脱敏（敏感键掩码/嵌套/非敏感保留）、json 导出可解析、csv 导出可解析、
|
||
# 导出元信息、网关端点。
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import io
|
||
import json
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.agent_core.audit_export import (
|
||
REDACTED, export_events, sanitize_event,
|
||
)
|
||
|
||
|
||
def _event(i: int = 1) -> dict:
|
||
return {"id": i, "at": "2026-08-01T00:00:00+08:00", "actor": "tester",
|
||
"category": "TEST", "action": f"test.{i}",
|
||
"target": {"type": "T"}, "power": "P0", "rationale": {},
|
||
"evidenceRefs": ["run-1"], "result": "SUCCESS",
|
||
"prevHash": "GENESIS", "hash": "0" * 64}
|
||
|
||
|
||
def test_sanitize_redacts_sensitive_keys():
|
||
"""敏感键值被掩码；非敏感值保留。"""
|
||
event = {
|
||
"id": 1,
|
||
"actor": "planner",
|
||
"rationale": {"confirmId": "abc", "token": "secret-token", "ok": True},
|
||
"meta": {"apiKey": "k-123", "engine": "CP"},
|
||
"list": [{"password": "p"}, {"note": "keep"}],
|
||
}
|
||
cleaned = sanitize_event(event)
|
||
assert cleaned["actor"] == "planner" # 非敏感保留
|
||
assert cleaned["rationale"]["confirmId"] == "abc"
|
||
assert cleaned["rationale"]["token"] == REDACTED
|
||
assert cleaned["meta"]["apiKey"] == REDACTED
|
||
assert cleaned["meta"]["engine"] == "CP"
|
||
assert cleaned["list"][0]["password"] == REDACTED
|
||
assert cleaned["list"][1]["note"] == "keep"
|
||
|
||
|
||
def test_export_json_parsable_and_complete():
|
||
"""json 导出可解析、事件完整、含元信息。"""
|
||
events = [_event(1), _event(2)]
|
||
body = export_events(events, fmt="json")
|
||
data = json.loads(body)
|
||
assert data["count"] == 2
|
||
assert len(data["events"]) == 2
|
||
assert "exportedAt" in data
|
||
assert data["events"][0]["action"] == "test.1"
|
||
|
||
|
||
def test_export_csv_parsable():
|
||
"""csv 导出可解析，表头含事件键。"""
|
||
events = [_event(1), _event(2)]
|
||
body = export_events(events, fmt="csv")
|
||
reader = list(csv.reader(io.StringIO(body)))
|
||
assert reader[0] == sorted(_event(1).keys()) # 表头
|
||
assert len(reader) == 3 # 表头 + 2 行
|
||
assert any("test.1" in row for row in reader[1:])
|
||
|
||
|
||
def test_sanitize_preserves_worldkey_and_similar():
|
||
"""精确匹配：worldKey/monkey 等含 'key' 子串的非敏感字段保留，apiKey/token 掩码。"""
|
||
event = {
|
||
"worldKey": "personal-1001",
|
||
"monkey": "banana",
|
||
"rationale": {"apiKey": "k-123", "token": "t-9", "worldKey": "w-1"},
|
||
"secret": "s",
|
||
}
|
||
cleaned = sanitize_event(event)
|
||
assert cleaned["worldKey"] == "personal-1001"
|
||
assert cleaned["monkey"] == "banana"
|
||
assert cleaned["rationale"]["worldKey"] == "w-1"
|
||
assert cleaned["rationale"]["apiKey"] == REDACTED
|
||
assert cleaned["rationale"]["token"] == REDACTED
|
||
assert cleaned["secret"] == REDACTED
|
||
|
||
|
||
def test_export_rejects_unknown_format():
|
||
"""未知格式拒绝。"""
|
||
with pytest.raises(ValueError):
|
||
export_events([_event()], fmt="xml")
|
||
|
||
|
||
def test_gateway_export_endpoint(monkeypatch, tmp_path):
|
||
"""网关导出端点：登录后返回 ok + 脱敏 body。"""
|
||
import server.gateway.app as gateway_module
|
||
from tests.auth_provider import install_test_auth
|
||
install_test_auth(monkeypatch, "tenant-export-test")
|
||
|
||
class _Store:
|
||
tenant_uuid = "platform"
|
||
world_key = "default"
|
||
data = {"auditEvents": [_event(1)]}
|
||
def next_id(self, _k): return 2
|
||
def save(self): pass
|
||
|
||
monkeypatch.setattr(gateway_module, "get_store", lambda: _Store())
|
||
client = TestClient(gateway_module.create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
resp = client.get("/api/gov/audit/export?format=json")
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()
|
||
assert data["ok"] is True and data["count"] == 1
|
||
payload = json.loads(data["body"])
|
||
assert payload["events"][0]["action"] == "test.1"
|
||
# csv 格式
|
||
resp2 = client.get("/api/gov/audit/export?format=csv")
|
||
assert resp2.status_code == 200
|