240 lines
7.9 KiB
Python
240 lines
7.9 KiB
Python
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()
|