aps-agent/tests/golden/test_audit_worm.py

# ============================================================
# WORM 归档模拟 + 完整性验证黄金测试（矩阵 104/112/116 的 WORM 剩余项）
# 覆盖：归档生成 manifest（sha256 + merkle root + 归档时间 + 覆盖范围）、
#       归档后写拒绝（只读位 + manifest 在册）、同月二次归档不改写旧件、
#       篡改检出（tampered expected/actual）、缺失/无清单显式报告、
#       merkle 复算校验、verify 汇总结构、网关端点接线（P0 verify / P1 run）。
# 说明：物理 WORM 介质（光盘/WORM 盘）为外部验收范围，本测试验证逻辑 WORM 语义。
# ============================================================
from __future__ import annotations

import hashlib
import json
import os
import stat
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.agent_core.audit_ledger import WORM_MANIFEST, WORM_NOTE, AnchorLedger
from server.agent_core.audit_merkle import merkle_root


def _event(i: int) -> dict:
    return {"id": i, "at": "2026-08-01T00:00:00+08:00", "actor": "tester",
            "category": "TEST", "action": f"test.{i}", "target": {"type": "TEST"},
            "power": "P0", "rationale": {}, "prevHash": "GENESIS", "hash": "0" * 64}


def _seed_ledger(ledger: AnchorLedger, n: int = 5) -> None:
    evs: list[dict] = []
    for i in range(n):
        evs.append(_event(i + 1))
        ledger.append(evs, reason="test")


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _manifest(ledger: AnchorLedger) -> dict:
    return json.loads((ledger.archive_dir / WORM_MANIFEST).read_text(encoding="utf-8"))


def test_archive_run_generates_worm_manifest(tmp_path: Path):
    """归档生成 WORM 清单：每文件 sha256 + merkleRoot + 归档时间 + 覆盖范围 + 只读位。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    assert r["archived"] == 3
    assert r["worm"]["sealed"] is True
    assert r["worm"]["mode"] == "worm-simulated"

    manifest = _manifest(ledger)
    assert manifest["mode"] == "worm-simulated"
    assert WORM_NOTE in manifest["note"]               # 文档化：物理介质为外部验收
    fname = Path(r["archive"]).name
    entry = manifest["files"][fname]
    assert entry["sha256"] == _sha256(Path(r["archive"]))
    assert entry["merkleRoot"]                          # merkle root 摘要
    assert entry["archivedAt"]                          # 归档时间
    assert entry["coverage"]["records"] == 3            # 覆盖范围
    assert entry["coverage"]["firstEventId"] and entry["coverage"]["lastEventId"]
    assert entry["readonly"] is True                    # WORM 只读位


def test_archive_file_write_refused_after_seal(tmp_path: Path):
    """归档后常规写路径拒绝改写：文件只读（直接写抛错）+ manifest 在册拒绝。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    archive = Path(r["archive"])
    # 常规文件写路径被只读位拒绝（Windows 抛 PermissionError）
    with pytest.raises((PermissionError, OSError)):
        archive.write_text(archive.read_text(encoding="utf-8") + "tamper\n", encoding="utf-8")
    # 账本写路径对已密封归档文件显式拒绝（manifest 在册 → ValueError）
    with pytest.raises(ValueError):
        ledger._guard_archive_sealed(archive)


def test_second_archive_same_month_never_rewrites_sealed(tmp_path: Path):
    """同月二次归档不改写已密封文件：改用 <month>.<n>.jsonl，旧件 sha256 不变。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger, n=5)
    r1 = ledger.archive_old(keep_last=2)
    first = Path(r1["archive"])
    first_sha = _sha256(first)
    # 追加新锚定记录后再次归档（同月）→ 新文件而非覆写旧件
    _seed_ledger(ledger, n=3)
    r2 = ledger.archive_old(keep_last=2)
    second = Path(r2["archive"])
    assert second.name != first.name
    assert second.name == f"{first.stem}.2.jsonl" or second.name.endswith(".2.jsonl")
    assert _sha256(first) == first_sha                  # 旧件字节级未变
    manifest = _manifest(ledger)
    assert first.name in manifest["files"] and second.name in manifest["files"]


def test_verify_archive_detects_tampering(tmp_path: Path):
    """篡改检出：改归档文件内容 → tampered {file, expected, actual}，ok=False。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    archive = Path(r["archive"])
    assert ledger.verify_archive()["ok"] is True

    os.chmod(archive, stat.S_IWRITE)                    # 模拟攻击者清除只读位
    archive.write_text(
        archive.read_text(encoding="utf-8")
        + json.dumps({"eventId": "EVIL", "root": "0" * 64, "count": 1,
                      "at": "2026-08-01 00:00", "reason": "tamper"}) + "\n",
        encoding="utf-8")
    result = ledger.verify_archive()
    assert result["ok"] is False
    assert result["summary"]["tampered"] == 1
    item = result["tampered"][0]
    assert item["file"] == archive.name
    assert item["expected"] == r["worm"]["entry"]["sha256"]
    assert item["actual"] == _sha256(archive)
    assert item["actual"] != item["expected"]


def test_verify_archive_reports_missing_and_no_manifest(tmp_path: Path):
    """缺失文件显式报告；无归档（无清单）给出确定性结构且 ok=True。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    archive = Path(r["archive"])
    os.chmod(archive, stat.S_IWRITE)                    # 清除只读位以便删除
    archive.unlink()
    result = ledger.verify_archive()
    assert result["ok"] is False
    assert result["summary"]["missing"] == 1
    assert result["missing"][0]["file"] == archive.name
    assert result["missing"][0]["expected"]             # 清单中的期望 sha256

    fresh = AnchorLedger("tenant-fresh", "fresh", ledger_dir=str(tmp_path / "other"))
    empty = fresh.verify_archive()
    assert empty["ok"] is True                          # 无归档 = 无篡改
    assert empty["summary"]["noManifest"] is True
    assert empty["manifest"] is None


def test_verify_archive_merkle_recompute_and_summary(tmp_path: Path):
    """merkle 复算一致 + verify 汇总字段齐全。"""
    ledger = AnchorLedger("tenant-w", "world-w", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    archive = Path(r["archive"])
    records = [json.loads(ln) for ln in
               archive.read_text(encoding="utf-8").splitlines() if ln.strip()]
    manifest = _manifest(ledger)
    assert manifest["files"][archive.name]["merkleRoot"] == merkle_root(records)

    result = ledger.verify_archive()
    assert result["ok"] is True
    assert result["mode"] == "worm-simulated"
    s = result["summary"]
    for key in ("files", "verified", "tampered", "missing", "merkleMismatch",
                "unsealed", "noManifest"):
        assert key in s
    assert s["files"] == 1 and s["verified"] == 1
    assert s["tampered"] == 0 and s["missing"] == 0 and s["merkleMismatch"] == 0
    assert result["tampered"] == [] and result["missing"] == []


class _Store:
    def __init__(self):
        self.tenant_uuid = "platform"
        self.world_key = "default"
        self.data = {"auditEvents": [
            {"id": 1, "at": "2026-08-01T00:00:00+08:00", "ts": "2026-08-01T00:00:00+08:00",
             "actor": "tester", "category": "TEST", "action": "test.one",
             "target": {"type": "TEST"}, "power": "P0", "rationale": {},
             "prevHash": "GENESIS", "hash": "0" * 64},
        ]}

    def next_id(self, _kind: str) -> int:
        return 2

    def save(self) -> None:
        pass


@pytest.fixture
def client(monkeypatch, tmp_path):
    import server.gateway.app as gateway_module
    from tests.auth_provider import install_test_auth
    install_test_auth(monkeypatch, "tenant-worm-api")
    store = _Store()
    monkeypatch.setattr(gateway_module, "get_store", lambda: store)
    c = TestClient(gateway_module.create_app())
    login = c.post("/api/auth/login", json={"username": "planner", "password": "test"})
    assert login.status_code == 200, login.text
    return c, store


def test_gov_archive_endpoints_roundtrip(client, tmp_path):
    """网关接线：/api/gov/archive/run 触发归档 + 生成 manifest；/verify 只读返回汇总明细。"""
    c, store = client
    os.environ["APS_AUDIT_LEDGER_DIR"] = str(tmp_path / "ledger")
    try:
        v0 = c.get("/api/gov/archive/verify")
        assert v0.status_code == 200
        assert v0.json()["mode"] == "worm-simulated"
        assert v0.json()["summary"]["noManifest"] is True

        for _ in range(3):                               # 锚定 3 条账本记录
            assert c.post("/api/gov/audit/anchor").status_code == 200
        run = c.post("/api/gov/archive/run?keep_last=1")
        assert run.status_code == 200, run.text
        body = run.json()
        assert body["archived"] == 2
        assert body["worm"]["sealed"] is True

        v = c.get("/api/gov/archive/verify")
        assert v.status_code == 200
        data = v.json()
        assert data["ok"] is True
        assert data["summary"]["verified"] == 1
        assert data["tampered"] == [] and data["missing"] == []
        assert data["summary"]["noManifest"] is False
    finally:
        os.environ.pop("APS_AUDIT_LEDGER_DIR", None)