from __future__ import annotations import base64 import copy import hashlib import json from datetime import datetime, timezone from pathlib import Path from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_audit.backend.db.models import AuditEvidenceBundle, AuditLog from govoplan_audit.backend.evidence_bundles import ( EvidenceBundleError, build_evidence_bundle, canonical_sha256, sanitize_audit_details, ) from govoplan_audit.backend.evidence_verifier import verify_evidence_bundle from govoplan_core.db.base import Base FIXTURES = Path(__file__).with_name("fixtures") NOW = datetime(2026, 8, 20, 10, 0, tzinfo=timezone.utc) def _record() -> AuditLog: item = AuditLog( id="audit-1", scope="tenant", tenant_id="tenant-1", user_id=None, api_key_id=None, action="case.decision.recorded", object_type="case", object_id="case-42", details={ "correlation_id": "trace-42", "policy_decision_ref": "policy:42", "source_ref": "cases:case-42:v7", "recipient_count": 2, "recipients": ["one@example.test", "two@example.test"], "message_body": "restricted body", "api_token": "secret-token", }, ) item.created_at = NOW item.updated_at = NOW return item def _bundle(*, references=(), signing_key_id=None, signing_private_key_path=None): return build_evidence_bundle( [_record()], bundle_id="bundle-1", generated_at=NOW, scope={"kind": "tenant", "tenant_id": "tenant-1"}, request={"record_ids": ["audit-1"], "selection_complete": True}, references=references, signing_key_id=signing_key_id, signing_private_key_path=signing_private_key_path, ) def test_bundle_is_deterministic_and_redacts_prohibited_evidence() -> None: first = _bundle() second = _bundle() assert first == second assert canonical_sha256(first) == canonical_sha256(second) details = first["records"][0]["details"] assert details["recipient_count"] == 2 assert details["recipients"] == {"redacted": True, "item_count": 2} assert details["message_body"] == {"redacted": True} assert details["api_token"] == {"redacted": True} encoded = json.dumps(first, sort_keys=True) assert "one@example.test" not in encoded assert "restricted body" not in encoded assert "secret-token" not in encoded assert verify_evidence_bundle(first)["status"] == "valid" def test_sanitizer_bounds_non_json_values_and_oversized_details() -> None: sanitized, paths = sanitize_audit_details({"custom": object(), "body": "secret"}) assert sanitized["custom"]["redacted"] is True assert sanitized["body"] == {"redacted": True} assert paths == ["details.body", "details.custom"] def test_reference_locators_cannot_embed_credentials_or_tokens() -> None: reference = { "reference_id": "decision-42", "kind": "module_evidence", "owner_module": "decisions", "locator": "https://user:password@example.test/evidence", } try: _bundle(references=[reference]) except EvidenceBundleError as exc: assert "appears to contain a secret" in str(exc) else: # pragma: no cover - explicit fail-closed assertion raise AssertionError("Secret-bearing evidence locators must be rejected") def test_verifier_distinguishes_missing_unverifiable_and_tampered_evidence() -> None: artifact = b"canonical external evidence" reference = { "reference_id": "decision-42", "kind": "module_evidence", "owner_module": "decisions", "locator": "decisions:decision-42:v3", "content_sha256": hashlib.sha256(artifact).hexdigest(), "required": True, } bundle = _bundle(references=[reference]) missing = verify_evidence_bundle(bundle) assert missing["status"] == "incomplete" assert missing["errors"][0]["code"] == "external_evidence_missing" valid = verify_evidence_bundle( bundle, external_evidence={"decision-42": artifact}, ) assert valid["status"] == "valid" mismatched = verify_evidence_bundle( bundle, external_evidence={"decision-42": b"changed"}, ) assert mismatched["status"] == "tampered" assert mismatched["errors"][0]["code"] == "external_evidence_hash_mismatch" unverifiable_reference = dict(reference) unverifiable_reference["content_sha256"] = None unverifiable = verify_evidence_bundle(_bundle(references=[unverifiable_reference])) assert unverifiable["status"] == "unverifiable" assert unverifiable["warnings"][1]["code"] == "reference_unverifiable" def test_verifier_detects_record_and_manifest_tampering() -> None: bundle = _bundle() changed = copy.deepcopy(bundle) changed["records"][0]["action"] = "case.decision.deleted" result = verify_evidence_bundle(changed) assert result["status"] == "tampered" assert {item["code"] for item in result["errors"]} == {"item_hash_mismatch"} def test_trusted_ed25519_signature_verifies_offline(tmp_path: Path) -> None: private_key = Ed25519PrivateKey.generate() private_path = tmp_path / "audit-evidence.pem" private_path.write_bytes( private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) ) public_key = base64.b64encode( private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) ).decode("ascii") bundle = _bundle( signing_key_id="institution-2026", signing_private_key_path=private_path, ) trusted = verify_evidence_bundle( bundle, trusted_keys={"institution-2026": public_key}, ) untrusted = verify_evidence_bundle(bundle) assert trusted["status"] == "valid" assert trusted["signature_state"] == "trusted" assert untrusted["status"] == "unverifiable" assert untrusted["signature_state"] == "unverifiable" def test_unsupported_version_is_reported_without_guessing() -> None: bundle = _bundle() bundle["version"] = "2.0" result = verify_evidence_bundle(bundle) assert result["status"] == "unsupported" assert result["errors"][0]["code"] == "unsupported_version" def test_supported_v1_compatibility_fixture_verifies_deterministically() -> None: payload = json.loads((FIXTURES / "evidence_bundle_v1.json").read_text(encoding="utf-8")) first = verify_evidence_bundle(payload) second = verify_evidence_bundle(payload) assert first == second assert first["status"] == "valid" def test_evidence_bundle_lifecycle_is_persisted() -> None: engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(bind=engine, tables=[AuditEvidenceBundle.__table__]) Session = sessionmaker(bind=engine) with Session() as session: row = AuditEvidenceBundle( id="bundle-1", scope="tenant", tenant_id="tenant-1", requested_by="operator-1", status="pending", request_payload={"record_ids": ["audit-1"]}, ) session.add(row) session.flush() assert row.status == "pending" row.status = "ready" row.bundle_payload = _bundle() row.bundle_sha256 = canonical_sha256(row.bundle_payload) row.record_count = 1 row.generated_at = NOW session.commit() restored = session.get(AuditEvidenceBundle, "bundle-1") assert restored is not None assert restored.status == "ready" assert restored.record_count == 1 assert restored.bundle_payload["version"] == "1.0" engine.dispose()