350 lines
15 KiB
Python
350 lines
15 KiB
Python
# ============================================================
|
||
# 加密归档模拟黄金测试（矩阵 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" |