from __future__ import annotations import base64 from copy import deepcopy from datetime import UTC, datetime import hashlib from importlib import metadata import json from pathlib import Path import sys import tempfile from types import SimpleNamespace import unittest from unittest import mock from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey META_ROOT = Path(__file__).resolve().parents[1] ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): if str(tools_root) not in sys.path: sys.path.insert(0, str(tools_root)) from govoplan_assessment.capability_fit import review_capability_fit # noqa: E402 from govoplan_assessment.evidence import ( # noqa: E402 canonical_bytes, canonical_sha256, collect_installed_composition, review_installed_composition, validate_payload, ) from tests.test_capability_fit_review import signed_catalog # noqa: E402 class CapabilityFitEvidenceTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.assessment = load_json("docs/capability-fit-current.json") cls.assessment_schema = load_json("docs/capability-fit.schema.json") cls.installed_schema = load_json( "docs/installed-composition-evidence.schema.json" ) cls.boundary_schema = load_json( "docs/capability-fit-boundary-evidence.schema.json" ) cls.authority_schema = load_json( "docs/capability-fit-proof-authority-keyring.schema.json" ) def test_matching_installed_composition_proves_versions_records_and_commits( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) self.assertEqual("current", report["status"]) self.assertTrue(report["proof_scope"]["installed_artifacts"]["valid"]) self.assertTrue(report["proof_scope"]["installed_record_integrity"]["valid"]) self.assertTrue(report["proof_scope"]["installed_source_provenance"]["valid"]) self.assertFalse(report["proof_scope"]["runtime_activation"]["checked"]) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertEqual( canonical_sha256(evidence), report["proof_scope"]["installed_artifacts"]["evidence_sha256"], ) def test_missing_extra_and_version_drift_have_deterministic_review_targets( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) evidence["artifacts"] = [ item for item in evidence["artifacts"] if item["package_name"] != "govoplan-campaign" ] access = next( item for item in evidence["artifacts"] if item["package_name"] == "govoplan-access" ) access["package_version"] = "9.9.9" evidence["artifacts"].append( artifact( package_name="govoplan-unassessed", version="1.0.0", module_id="unassessed", commit="f" * 40, ) ) first = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) second = self.review( catalog=catalog, keyring=keyring, installed_evidence=deepcopy(evidence), ) self.assertEqual("review_required", first["status"]) self.assertEqual( json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True) ) codes = {item["code"] for item in first["findings"]} self.assertIn("installed_distribution_missing", codes) self.assertIn("installed_distribution_extra", codes) self.assertIn("installed_distribution_version_mismatch", codes) targets = {item["id"] for item in first["review_targets"]} self.assertIn("composition.campaigns", targets) self.assertIn("composition.access", targets) self.assertIn("assessment.installed_composition", targets) self.assertTrue(first["proof_scope"]["release_metadata"]["valid"]) self.assertFalse(first["proof_scope"]["installed_artifacts"]["valid"]) def test_editable_source_and_partial_record_are_not_immutable_proof(self) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) campaign = next( item for item in evidence["artifacts"] if item["package_name"] == "govoplan-campaign" ) campaign["source_provenance"] = {"kind": "editable-local"} campaign["record_integrity"] = { **campaign["record_integrity"], "status": "partial", "unverifiable_file_count": 1, } report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) codes = {item["code"] for item in report["findings"]} self.assertIn("installed_source_mutable", codes) self.assertIn("installed_record_integrity_unverified", codes) self.assertFalse(report["proof_scope"]["installed_source_provenance"]["valid"]) self.assertFalse(report["proof_scope"]["installed_record_integrity"]["valid"]) def test_malformed_or_wrongly_bound_installed_evidence_fails_closed(self) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) evidence["assessment_id"] = "another-assessment" evidence["artifacts"].append(deepcopy(evidence["artifacts"][0])) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) self.assertEqual("blocked", report["status"]) codes = {item["code"] for item in report["findings"]} self.assertIn("installed_evidence_assessment_mismatch", codes) self.assertIn("installed_distribution_duplicate", codes) self.assertFalse(report["proof_scope"]["installed_artifacts"]["valid"]) self.assertIn( "composition.core", {item["id"] for item in report["review_targets"]} ) self.assertEqual( 0, report["proof_scope"]["installed_source_provenance"][ "mutable_distribution_count" ], ) def test_schema_failure_does_not_echo_untrusted_evidence_values(self) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) evidence["artifacts"][0]["package_name"] = "secret-user@private-host/path" report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) encoded = json.dumps(report, sort_keys=True) self.assertEqual("blocked", report["status"]) self.assertIn("installed_evidence_schema", encoded) self.assertNotIn("secret-user", encoded) self.assertNotIn("private-host", encoded) def test_missing_catalog_package_mapping_cannot_pass_vacuously(self) -> None: evidence = matching_installed_evidence(self.assessment) result = review_installed_composition( assessment=self.assessment, catalog_entries={}, selected_versions={}, selected_commits={}, evidence=evidence, schema=self.installed_schema, ) codes = {item.code for item in result.findings} self.assertIn("installed_expected_catalog_entry_missing", codes) self.assertIn("installed_expected_composition_empty", codes) self.assertTrue(result.proof_scope["installed_artifacts"]["checked"]) self.assertFalse(result.proof_scope["installed_artifacts"]["valid"]) def test_collector_verifies_record_and_redacts_direct_url(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) distribution, payload_file = create_distribution(root) with mock.patch.object(sys, "path", [str(root), *sys.path]): evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), distributions=[distribution], ) self.assertEqual( (), validate_payload(payload=evidence, schema=self.installed_schema) ) observed = evidence["artifacts"][0] self.assertEqual("editable-local", observed["source_provenance"]["kind"]) self.assertEqual("verified", observed["record_integrity"]["status"]) encoded = json.dumps(evidence, sort_keys=True) self.assertNotIn(str(root), encoded) self.assertNotIn("file://", encoded) payload_file.write_text("tampered\n", encoding="utf-8") with mock.patch.object(sys, "path", [str(root), *sys.path]): tampered = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), distributions=[distribution], ) self.assertEqual( "mismatch", tampered["artifacts"][0]["record_integrity"]["status"] ) self.assertEqual( 1, tampered["artifacts"][0]["record_integrity"]["mismatched_file_count"], ) def test_entry_point_limit_is_explicit_instead_of_silent_truncation(self) -> None: distribution = FakeDistribution( entry_points=[FakeEntryPoint(name=f"module-{index}") for index in range(17)] ) evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), distributions=[distribution], ) self.assertEqual(16, len(evidence["artifacts"][0]["modules"])) self.assertIn( { "code": "module-entry-point-limit-exceeded", "package_name": "govoplan-many", }, evidence["collection_issues"], ) def test_tied_distribution_rows_are_canonically_ordered(self) -> None: first_distribution = FakeDistribution( entry_points=[FakeEntryPoint(name="module-z")], direct_url={"dir_info": {"editable": True}, "url": "file:///redacted-a"}, ) second_distribution = FakeDistribution( entry_points=[FakeEntryPoint(name="module-a")], direct_url={"url": "https://redacted.invalid/archive"}, ) collected_at = datetime(2026, 7, 22, 12, tzinfo=UTC) forward = collect_installed_composition( assessment=self.assessment, collected_at=collected_at, distributions=[first_distribution, second_distribution], ) reversed_order = collect_installed_composition( assessment=self.assessment, collected_at=collected_at, distributions=[second_distribution, first_distribution], ) self.assertEqual( json.dumps(forward, sort_keys=True), json.dumps(reversed_order, sort_keys=True), ) def test_signed_scope_authority_can_validate_only_its_external_claim(self) -> None: catalog, keyring = signed_catalog(self.assessment) installed = matching_installed_evidence(self.assessment) proof, authority = signed_boundary_evidence( assessment=self.assessment, installed=installed, claims=[boundary_claim("target_environment", "passed")], allowed_scopes=["target_environment"], ) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) self.assertEqual("current", report["status"]) self.assertTrue(report["proof_scope"]["target_environment"]["valid"]) self.assertFalse(report["proof_scope"]["external_providers"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) def test_unauthorized_production_claim_keeps_every_claim_unchecked(self) -> None: catalog, keyring = signed_catalog(self.assessment) installed = matching_installed_evidence(self.assessment) proof, authority = signed_boundary_evidence( assessment=self.assessment, installed=installed, claims=[ boundary_claim("target_environment", "passed"), boundary_claim("production_approval", "approved"), ], allowed_scopes=["target_environment"], ) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) self.assertEqual("blocked", report["status"]) self.assertIn( "boundary_authority_untrusted", {item["code"] for item in report["findings"]}, ) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) def test_boundary_semantics_interval_and_duplicates_fail_closed(self) -> None: catalog, keyring = signed_catalog(self.assessment) installed = matching_installed_evidence(self.assessment) cases = [] semantic = [boundary_claim("production_approval", "passed")] cases.append(("semantics", semantic, None)) duplicate = [ boundary_claim("target_environment", "passed"), boundary_claim("target_environment", "failed"), ] cases.append(("duplicate", duplicate, None)) cases.append( ( "interval", [boundary_claim("target_environment", "passed")], ("2026-07-24T00:00:00Z", "2026-07-23T00:00:00Z"), ) ) for label, claims, interval in cases: with self.subTest(label=label): proof, authority = signed_boundary_evidence( assessment=self.assessment, installed=installed, claims=claims, allowed_scopes=[ "target_environment", "production_approval", ], interval=interval, ) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) self.assertEqual("blocked", report["status"]) self.assertFalse( report["proof_scope"]["production_approval"]["checked"] ) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) def review( self, *, catalog: dict[str, object], keyring: dict[str, object], installed_evidence: dict[str, object] | None = None, boundary_evidence: dict[str, object] | None = None, authority_keyring: dict[str, object] | None = None, ) -> dict[str, object]: return review_capability_fit( assessment=deepcopy(self.assessment), schema=self.assessment_schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, installed_evidence=installed_evidence, installed_evidence_schema=self.installed_schema if installed_evidence is not None else None, boundary_evidence=boundary_evidence, boundary_evidence_schema=self.boundary_schema if boundary_evidence is not None else None, boundary_authority_keyring=authority_keyring, boundary_authority_keyring_schema=self.authority_schema if boundary_evidence is not None else None, verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), ) def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]: return { "$schema": "./installed-composition-evidence.schema.json", "schema_version": "0.1.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], "collected_at": "2026-07-22T12:00:00Z", "scope": "current-python-environment.govoplan-distributions", "artifacts": [ artifact( package_name=str(component["repository"]), version=str(component["manifest_version"]), module_id=str(component["module_id"]), commit=(str(component["commit"]) + "0" * 64)[:40], ) for component in assessment["composition"] if component["enabled"] is True ], "collection_issues": [], } def artifact( *, package_name: str, version: str, module_id: str, commit: str ) -> dict[str, object]: return { "package_name": package_name, "package_version": version, "modules": [] if module_id == "core" else [{"module_id": module_id, "manifest_version": version}], "source_provenance": {"kind": "vcs-commit", "commit": commit}, "record_integrity": { "status": "verified", "hashed_file_count": 1, "permitted_unhashed_file_count": 1, "unverifiable_file_count": 0, "missing_file_count": 0, "mismatched_file_count": 0, }, } def boundary_claim(scope: str, result: str) -> dict[str, object]: return { "scope": scope, "result": result, "subject_id": f"subject:{scope}", "control_ids": [f"control:{scope}"], "artifacts": [ { "artifact_id": f"result:{scope}", "sha256": hashlib.sha256(scope.encode()).hexdigest(), } ], } def signed_boundary_evidence( *, assessment: dict[str, object], installed: dict[str, object], claims: list[dict[str, object]], allowed_scopes: list[str], interval: tuple[str, str] | None = None, ) -> tuple[dict[str, object], dict[str, object]]: private_key = Ed25519PrivateKey.generate() public_key = base64.b64encode( private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) ).decode("ascii") issued_at, expires_at = interval or ( "2026-07-22T00:00:00Z", "2026-07-24T00:00:00Z", ) proof: dict[str, object] = { "$schema": "./capability-fit-boundary-evidence.schema.json", "schema_version": "0.1.0", "evidence_kind": "govoplan.capability-fit-boundary-proof", "proof_id": "proof:test", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], "installed_evidence_sha256": canonical_sha256(installed), "issued_at": issued_at, "expires_at": expires_at, "claims": claims, } proof["signatures"] = [ { "algorithm": "ed25519", "key_id": "authority:test", "value": base64.b64encode(private_key.sign(canonical_bytes(proof))).decode( "ascii" ), } ] keyring = { "$schema": "./capability-fit-proof-authority-keyring.schema.json", "schema_version": "0.1.0", "purpose": "govoplan.capability-fit-proof-authorities", "keys": [ { "key_id": "authority:test", "status": "active", "public_key": public_key, "allowed_scopes": allowed_scopes, "not_before": "2026-07-01T00:00:00Z", "not_after": "2026-08-01T00:00:00Z", } ], } return proof, keyring def create_distribution( root: Path, ) -> tuple[metadata.PathDistribution, Path]: payload_file = root / "govoplan_demo.py" payload_file.write_text("value = 1\n", encoding="utf-8") dist_info = root / "govoplan_demo-1.0.0.dist-info" dist_info.mkdir() metadata_file = dist_info / "METADATA" metadata_file.write_text( "Metadata-Version: 2.1\nName: govoplan-demo\nVersion: 1.0.0\n", encoding="utf-8", ) direct_url_file = dist_info / "direct_url.json" direct_url_file.write_text( json.dumps( { "url": f"file://{root}/private-user/govoplan-demo", "dir_info": {"editable": True}, } ), encoding="utf-8", ) record_file = dist_info / "RECORD" rows = [] for path in (payload_file, metadata_file, direct_url_file): relative = path.relative_to(root).as_posix() content = path.read_bytes() digest = ( base64.urlsafe_b64encode(hashlib.sha256(content).digest()) .decode("ascii") .rstrip("=") ) rows.append(f"{relative},sha256={digest},{len(content)}") rows.append(f"{record_file.relative_to(root).as_posix()},,") record_file.write_text("\n".join(rows) + "\n", encoding="utf-8") return metadata.PathDistribution(dist_info), payload_file class FakeEntryPoint: group = "govoplan.modules" value = "fixture:get_manifest" def __init__(self, *, name: str) -> None: self.name = name def load(self): return lambda: SimpleNamespace(id=self.name, version="1.0.0") class FakeDistribution: version = "1.0.0" files = () def __init__( self, *, entry_points: list[FakeEntryPoint], direct_url: dict[str, object] | None = None, ) -> None: self.metadata = {"Name": "govoplan-many"} self.entry_points = entry_points self.direct_url = direct_url def read_text(self, filename: str): return ( json.dumps(self.direct_url) if filename == "direct_url.json" and self.direct_url else None ) def load_json(relative_path: str) -> dict[str, object]: return json.loads((META_ROOT / relative_path).read_text(encoding="utf-8")) if __name__ == "__main__": unittest.main()