aps-agent/tests/golden/test_audit_encryption.py

# ============================================================
# 加密归档模拟黄金测试（矩阵 104/116 的加密剩余项）
# 覆盖：AES-256-GCM 加密归档往返（env 原始密钥 / 密钥文件口令 scrypt 派生）、
#       manifest ciphertextHash + plaintextHash 双层校验、
#       解密校验（显式密钥）、密钥缺失显式失败（不静默）、错误密钥显式失败、
#       密文篡改检出、WORM manifest 兼容（不加密行为不变 / encrypt=False 强制明文）、
#       encrypt=True 无密钥 fail-closed、同月二次加密归档、网关端点接线。
# 说明：真实密钥管理（HSM/KMS、密钥轮换）为外部验收范围，本地为 env/密钥文件密钥。
# ============================================================
from __future__ import annotations

import json
import os
import stat
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.agent_core.audit_ledger import (
    ENCRYPTED_FORMAT,
    ENCRYPTION_ENV_KEY,
    ENCRYPTION_ENV_KEY_FILE,
    ENCRYPTION_KDF,
    ENCRYPTION_NOTE,
    ENCRYPTION_SCHEME,
    WORM_MANIFEST,
    AnchorLedger,
    _decrypt_bytes,
)
from server.agent_core.audit_merkle import merkle_root

RAW_KEY = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff"
WRONG_KEY = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
PASSPHRASE = "s3cret-passphrase-for-audit-archive"


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 _manifest(ledger: AnchorLedger) -> dict:
    return json.loads((ledger.archive_dir / WORM_MANIFEST).read_text(encoding="utf-8"))


def _entry(ledger: AnchorLedger, archive_name: str) -> dict:
    return _manifest(ledger)["files"][archive_name]


# ---------- 加密归档往返 ----------

def test_encrypt_archive_roundtrip_env_key(tmp_path: Path, monkeypatch):
    """env 原始密钥（64 hex）：归档文件为加密信封，manifest 双层哈希 + 算法/版本标记，
    解密校验通过。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-e", "world-e", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    assert r["archived"] == 3
    assert r["encryption"]["enabled"] is True
    assert r["encryption"]["scheme"] == ENCRYPTION_SCHEME

    archive = Path(r["archive"])
    envelope = json.loads(archive.read_text(encoding="utf-8"))
    assert envelope["format"] == ENCRYPTED_FORMAT
    assert envelope["scheme"] == ENCRYPTION_SCHEME
    assert envelope["ciphertext"]                      # 非明文（无 eventId 顶层字段）
    assert "eventId" not in envelope

    entry = _entry(ledger, archive.name)
    assert entry["encryption"]["enabled"] is True
    assert entry["encryption"]["scheme"] == ENCRYPTION_SCHEME
    assert entry["encryption"]["kdf"] == "none"        # 原始密钥不经 KDF
    assert entry["ciphertextHash"] == entry["sha256"]  # 密文文件哈希
    assert entry["plaintextHash"]                      # 明文哈希
    assert entry["ciphertextHash"] != entry["plaintextHash"]

    v = ledger.verify_archive()
    assert v["ok"] is True
    assert v["summary"]["verified"] == 1
    assert v["summary"]["decryptFailed"] == 0
    assert v["encryption"]["enabled"] is True and v["encryption"]["count"] == 1


def test_encrypt_archive_roundtrip_key_file_passphrase(tmp_path: Path, monkeypatch):
    """密钥文件 + 口令：kdf=scrypt；清除 env 后显式传密钥仍可解密校验。"""
    key_file = tmp_path / "audit.key"
    key_file.write_text(PASSPHRASE + "\n", encoding="utf-8")
    monkeypatch.setenv(ENCRYPTION_ENV_KEY_FILE, str(key_file))
    monkeypatch.delenv(ENCRYPTION_ENV_KEY, raising=False)
    ledger = AnchorLedger("tenant-p", "world-p", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    assert r["encryption"]["enabled"] is True
    entry = _entry(ledger, Path(r["archive"]).name)
    assert entry["encryption"]["kdf"] == ENCRYPTION_KDF  # scrypt 派生

    monkeypatch.delenv(ENCRYPTION_ENV_KEY_FILE)
    v_missing = ledger.verify_archive()
    assert v_missing["ok"] is False                     # 无密钥 → 显式失败
    assert v_missing["summary"]["decryptFailed"] == 1
    assert v_missing["decryptFailed"][0]["reason"] == "encryption-key-missing"

    v_explicit = ledger.verify_archive(key=PASSPHRASE.encode("utf-8"))
    assert v_explicit["ok"] is True
    assert v_explicit["summary"]["verified"] == 1


# ---------- 解密校验 / 显式失败 ----------

def test_verify_explicit_key_and_merkle_recompute(tmp_path: Path, monkeypatch):
    """显式密钥解密 + 明文 JSONL 复算 merkle root 与清单一致（双层校验闭环）。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-d", "world-d", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    archive = Path(r["archive"])
    monkeypatch.delenv(ENCRYPTION_ENV_KEY)

    envelope = json.loads(archive.read_text(encoding="utf-8"))
    plaintext = _decrypt_bytes(envelope, bytes.fromhex(RAW_KEY))
    import hashlib
    entry = _entry(ledger, archive.name)
    assert hashlib.sha256(plaintext).hexdigest() == entry["plaintextHash"]
    records = [json.loads(ln) for ln in plaintext.decode("utf-8").splitlines() if ln.strip()]
    assert len(records) == 3
    assert entry["merkleRoot"] == merkle_root(records)
    assert entry["coverage"]["records"] == 3

    v = ledger.verify_archive(key=bytes.fromhex(RAW_KEY))
    assert v["ok"] is True
    assert v["summary"]["verified"] == 1
    assert v["summary"]["merkleMismatch"] == 0


def test_verify_missing_key_fails_explicitly(tmp_path: Path, monkeypatch):
    """密钥缺失：verify 显式失败（decryptFailed=1），不静默视为通过。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-m", "world-m", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    assert r["encryption"]["enabled"] is True

    monkeypatch.delenv(ENCRYPTION_ENV_KEY)
    v = ledger.verify_archive()
    assert v["ok"] is False
    assert v["summary"]["decryptFailed"] == 1
    item = v["decryptFailed"][0]
    assert item["file"] == Path(r["archive"]).name
    assert item["reason"] == "encryption-key-missing"
    assert ENCRYPTION_NOTE in item["note"]             # 文档化：真实密钥管理为外部项
    assert v["summary"]["verified"] == 0               # 未静默计数为通过


def test_verify_wrong_key_fails_explicitly(tmp_path: Path, monkeypatch):
    """错误密钥：解密失败（AES-GCM tag 校验失败）→ decryptFailed，显式报告。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-wk", "world-wk", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    ledger.archive_old(keep_last=2)

    v = ledger.verify_archive(key=bytes.fromhex(WRONG_KEY))
    assert v["ok"] is False
    assert v["summary"]["decryptFailed"] == 1
    assert v["decryptFailed"][0]["reason"] == "decryption-failed"
    assert v["summary"]["tampered"] == 0


# ---------- 篡改检出 ----------

def test_verify_detects_ciphertext_tampering(tmp_path: Path, monkeypatch):
    """密文篡改：改信封文件字节 → ciphertext-hash-mismatch（tampered expected/actual）。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-ct", "world-ct", 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") + "{\"evil\":true}\n",
                       encoding="utf-8")
    v = ledger.verify_archive()
    assert v["ok"] is False
    assert v["summary"]["tampered"] == 1
    item = v["tampered"][0]
    assert item["reason"] == "ciphertext-hash-mismatch"
    entry = _entry(ledger, archive.name)
    assert item["expected"] == entry["ciphertextHash"]
    assert item["actual"] != item["expected"]


def test_verify_encrypted_archive_missing_file(tmp_path: Path, monkeypatch):
    """加密归档文件缺失 → missing 显式报告（与 WORM 语义一致）。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-ms", "world-ms", 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()
    v = ledger.verify_archive()
    assert v["ok"] is False
    assert v["summary"]["missing"] == 1
    assert v["missing"][0]["file"] == archive.name


# ---------- WORM manifest 兼容 ----------

def test_plaintext_behavior_unchanged_when_no_key(tmp_path: Path, monkeypatch):
    """不配置密钥：行为与 round-42 WORM 完全一致（encryption.enabled=False，
    sha256 == ciphertextHash == plaintextHash == 文件哈希）。"""
    monkeypatch.delenv(ENCRYPTION_ENV_KEY, raising=False)
    monkeypatch.delenv(ENCRYPTION_ENV_KEY_FILE, raising=False)
    ledger = AnchorLedger("tenant-pc", "world-pc", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2)
    assert r["encryption"]["enabled"] is False
    archive = Path(r["archive"])
    entry = _entry(ledger, archive.name)
    import hashlib
    file_sha = hashlib.sha256(archive.read_bytes()).hexdigest()
    assert entry["sha256"] == file_sha
    assert entry["ciphertextHash"] == file_sha
    assert entry["plaintextHash"] == file_sha
    assert entry["encryption"]["enabled"] is False
    records = [json.loads(ln) for ln in archive.read_text(encoding="utf-8").splitlines()
               if ln.strip()]                          # 明文 JSONL 仍可直读
    assert len(records) == 3
    assert ledger.verify_archive()["ok"] is True
    assert ledger.verify_archive()["encryption"]["enabled"] is False


def test_encrypt_true_without_key_fails_closed(tmp_path: Path, monkeypatch):
    """encrypt=True 但未配置密钥 → ValueError（fail-closed，不静默降级明文）。"""
    monkeypatch.delenv(ENCRYPTION_ENV_KEY, raising=False)
    monkeypatch.delenv(ENCRYPTION_ENV_KEY_FILE, raising=False)
    ledger = AnchorLedger("tenant-fc", "world-fc", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    with pytest.raises(ValueError, match="APS_AUDIT_ENCRYPT_KEY"):
        ledger.archive_old(keep_last=2, encrypt=True)


def test_encrypt_false_with_key_forces_plaintext(tmp_path: Path, monkeypatch):
    """配置了密钥但 encrypt=False：强制明文归档，WORM 清单加密标记为关闭。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-fp", "world-fp", ledger_dir=str(tmp_path))
    _seed_ledger(ledger)
    r = ledger.archive_old(keep_last=2, encrypt=False)
    assert r["encryption"]["enabled"] is False
    archive = Path(r["archive"])
    entry = _entry(ledger, archive.name)
    assert entry["encryption"]["enabled"] is False
    records = [json.loads(ln) for ln in archive.read_text(encoding="utf-8").splitlines()
               if ln.strip()]
    assert len(records) == 3                            # 明文 JSONL
    assert ledger.verify_archive()["ok"] is True


def test_second_encrypted_archive_same_month_new_file(tmp_path: Path, monkeypatch):
    """同月二次加密归档：不改写已密封文件，改用 <month>.<n>.jsonl，两件均在册。"""
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    ledger = AnchorLedger("tenant-2nd", "world-2nd", ledger_dir=str(tmp_path))
    _seed_ledger(ledger, n=5)
    r1 = ledger.archive_old(keep_last=2)
    first = Path(r1["archive"])
    _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")
    manifest = _manifest(ledger)
    assert first.name in manifest["files"] and second.name in manifest["files"]
    assert manifest["files"][first.name]["encryption"]["enabled"] is True
    assert manifest["files"][second.name]["encryption"]["enabled"] is True
    assert ledger.verify_archive()["ok"] is True
    assert ledger.verify_archive()["summary"]["verified"] == 2


# ---------- 网关接线（env 密钥，无需改网关代码） ----------

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-enc-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_encrypted_roundtrip(client, tmp_path, monkeypatch):
    """网关接线：配置 env 密钥 → /archive/run 产出加密归档，/archive/verify 通过；
    撤密钥 → verify 显式 decryptFailed（fail-closed 语义透出 API）。"""
    c, _ = client
    monkeypatch.setenv("APS_AUDIT_LEDGER_DIR", str(tmp_path / "ledger"))
    monkeypatch.setenv(ENCRYPTION_ENV_KEY, RAW_KEY)
    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["encryption"]["enabled"] is True
    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["summary"]["decryptFailed"] == 0

    monkeypatch.delenv(ENCRYPTION_ENV_KEY)              # 密钥缺失 → 显式失败
    v2 = c.get("/api/gov/archive/verify")
    assert v2.status_code == 200
    d2 = v2.json()
    assert d2["ok"] is False
    assert d2["summary"]["decryptFailed"] == 1
    assert d2["decryptFailed"][0]["reason"] == "encryption-key-missing"