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 subprocess 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 ( # noqa: E402 render_review, review_capability_fit, ) from govoplan_assessment.evidence import ( # noqa: E402 _bounded_file_sha256, 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_local_consistency_not_origin( 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.assertEqual( "local-pep610-metadata", report["proof_scope"]["installed_source_provenance"]["basis"], ) self.assertFalse( report["proof_scope"]["installed_source_provenance"][ "release_origin_bound" ] ) self.assertFalse(report["proof_scope"]["installed_release_origin"]["checked"]) self.assertIsNone(report["proof_scope"]["installed_release_origin"]["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_catalog_trust_blocker_invalidates_dependent_evidence_scopes( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) catalog["signatures"][0]["value"] = "AAAA" 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("blocked", report["status"]) self.assertFalse( report["proof_scope"]["catalog_signature_and_keyring"]["valid"] ) self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"]) self.assertIsNone(report["proof_scope"]["installed_artifacts"]["valid"]) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertIsNone(report["proof_scope"]["target_environment"]["valid"]) self.assertIn( "installed_catalog_trust_unavailable", {item["code"] for item in report["findings"]}, ) 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"] = { "basis": "local-pep610-metadata", "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, observation_mode="direct_local", verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), catalog_trusted=True, ) 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_imported_unsigned_installed_evidence_is_comparison_only(self) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, installed_evidence_mode="imported_unsigned", ) installed = report["proof_scope"]["installed_artifacts"] self.assertEqual("blocked", report["status"]) self.assertFalse(installed["checked"]) self.assertIsNone(installed["valid"]) self.assertTrue(installed["comparison_valid"]) self.assertEqual( "unsigned_import", report["proof_scope"]["installed_evidence_observation"]["freshness"], ) self.assertIn( "installed_evidence_unsigned_import", {item["code"] for item in report["findings"]}, ) def test_direct_installed_evidence_must_be_fresh_and_not_from_future(self) -> None: catalog, keyring = signed_catalog(self.assessment) cases = ( ("2026-07-23T11:54:59Z", "installed_evidence_stale", "stale"), ("2026-07-23T12:00:31Z", "installed_evidence_from_future", "future"), ) for collected_at, expected_code, expected_freshness in cases: with self.subTest(collected_at=collected_at): evidence = matching_installed_evidence(self.assessment) evidence["collected_at"] = collected_at report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) self.assertEqual("blocked", report["status"]) self.assertFalse( report["proof_scope"]["installed_artifacts"]["checked"] ) self.assertIn( expected_code, {item["code"] for item in report["findings"]}, ) self.assertEqual( expected_freshness, report["proof_scope"]["installed_evidence_observation"][ "freshness" ], ) def test_forged_verified_record_counters_fail_semantic_validation(self) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) evidence["artifacts"][0]["record_integrity"].update( hashed_file_count=0, mismatched_file_count=1, ) relaxed_schema = deepcopy(self.installed_schema) relaxed_schema["$defs"]["record_integrity"].pop("allOf") report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, installed_schema=relaxed_schema, ) self.assertEqual("review_required", report["status"]) self.assertIn( "installed_record_integrity_invalid", {item["code"] for item in report["findings"]}, ) self.assertFalse(report["proof_scope"]["installed_record_integrity"]["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( "local-pep610-metadata", observed["source_provenance"]["basis"], ) 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_collector_detects_payload_declared_in_record_but_missing(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) distribution, payload_file = create_distribution(root) payload_file.unlink() 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], ) integrity = evidence["artifacts"][0]["record_integrity"] self.assertEqual("mismatch", integrity["status"]) self.assertEqual(1, integrity["missing_file_count"]) def test_collector_rejects_duplicate_record_declarations(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) distribution, _ = create_distribution(root) record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" rows = record_path.read_text(encoding="utf-8").splitlines() record_path.write_text( "\n".join((*rows, rows[0])) + "\n", encoding="utf-8", ) 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], ) integrity = evidence["artifacts"][0]["record_integrity"] self.assertEqual("partial", integrity["status"]) self.assertGreater(integrity["unverifiable_file_count"], 0) def test_collector_rejects_malformed_record_declarations(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) distribution, _ = create_distribution(root) record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" record_path.write_text( record_path.read_text(encoding="utf-8") + "malformed,row\n", encoding="utf-8", ) 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], ) integrity = evidence["artifacts"][0]["record_integrity"] self.assertEqual("partial", integrity["status"]) self.assertGreater(integrity["unverifiable_file_count"], 0) def test_collector_verifies_declared_record_sizes(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) distribution, _ = create_distribution(root) record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" rows = record_path.read_text(encoding="utf-8").splitlines() path, digest, size = rows[0].split(",") rows[0] = f"{path},{digest},{int(size) + 1}" record_path.write_text("\n".join(rows) + "\n", encoding="utf-8") 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], ) integrity = evidence["artifacts"][0]["record_integrity"] self.assertEqual("mismatch", integrity["status"]) self.assertEqual(1, integrity["mismatched_file_count"]) def test_collector_rejects_symlinked_distribution_metadata_root(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) create_distribution(root) metadata_root = root / "govoplan_demo-1.0.0.dist-info" real_metadata_root = root / "real-govoplan-demo.dist-info" metadata_root.rename(real_metadata_root) metadata_root.symlink_to(real_metadata_root, target_is_directory=True) distribution = metadata.PathDistribution(metadata_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], ) integrity = evidence["artifacts"][0]["record_integrity"] self.assertEqual("unavailable", integrity["status"]) self.assertEqual(0, integrity["hashed_file_count"]) def test_collector_accepts_real_core_wheel_payload_scripts_data_and_pyc( self, ) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) wheels = root / "wheels" prefix = root / "prefix" wheels.mkdir() build_python = ( Path(sys.base_prefix) / "bin" / (f"python{sys.version_info.major}.{sys.version_info.minor}") ) subprocess.run( ( str(build_python), "-m", "pip", "wheel", "--no-deps", "--no-build-isolation", "--wheel-dir", str(wheels), str(META_ROOT.parent / "govoplan-core"), ), check=True, capture_output=True, text=True, ) wheel = next(wheels.glob("govoplan_core-*.whl")) subprocess.run( ( str(build_python), "-m", "pip", "install", "--no-deps", "--ignore-installed", "--prefix", str(prefix), str(wheel), ), check=True, capture_output=True, text=True, ) site_packages = next(prefix.glob("lib/python*/site-packages")) dist_info = next(site_packages.glob("govoplan_core-*.dist-info")) distribution = metadata.PathDistribution(dist_info) record_paths = {str(item) for item in distribution.files or ()} self.assertTrue( any( path.startswith("../../../govoplan_core_runtime/") for path in record_paths ) ) self.assertTrue( any(path.startswith("../../../bin/govoplan-") for path in record_paths) ) with ( mock.patch.object(sys, "path", [str(site_packages), *sys.path]), mock.patch.object(sys, "prefix", str(prefix)), ): evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), distributions=[distribution], ) record = evidence["artifacts"][0]["record_integrity"] self.assertEqual("verified", record["status"]) self.assertGreater(record["hashed_file_count"], 0) self.assertGreater(record["permitted_unhashed_file_count"], 0) self.assertGreater(record["generated_unhashed_file_count"], 0) self.assertEqual(0, record["unverifiable_file_count"]) self.assertEqual( (), validate_payload(payload=evidence, schema=self.installed_schema) ) def test_record_traversal_and_leaf_symlinks_are_never_hashed(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: prefix = Path(temp_dir) / "prefix" site_packages = prefix / "lib" / "python-test" / "site-packages" site_packages.mkdir(parents=True) traversal = create_traversal_distribution(prefix, site_packages) normal, payload = create_distribution(site_packages) outside = Path(temp_dir) / "outside.py" outside.write_text(payload.read_text(encoding="utf-8"), encoding="utf-8") payload.unlink() payload.symlink_to(outside) with ( mock.patch.object(sys, "path", [str(site_packages), *sys.path]), mock.patch.object(sys, "prefix", str(prefix)), ): evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), distributions=[traversal, normal], ) records = { item["package_name"]: item["record_integrity"] for item in evidence["artifacts"] } self.assertEqual("partial", records["govoplan-traversal"]["status"]) self.assertGreaterEqual( records["govoplan-traversal"]["unverifiable_file_count"], 2 ) self.assertEqual("partial", records["govoplan-demo"]["status"]) self.assertGreater(records["govoplan-demo"]["unverifiable_file_count"], 0) def test_bounded_hash_stops_at_opened_file_snapshot(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: path = Path(temp_dir) / "growing.bin" path.write_bytes(b"0123456789") actual_stat = path.stat() initial_stat = SimpleNamespace(st_mode=actual_stat.st_mode, st_size=4) final_stat = SimpleNamespace(st_mode=actual_stat.st_mode, st_size=10) with mock.patch( "govoplan_assessment.evidence.os.fstat", side_effect=(initial_stat, final_stat), ): _, hashed_bytes, exceeded = _bounded_file_sha256(path, max_bytes=8) self.assertEqual(4, hashed_bytes) self.assertTrue(exceeded) def test_collector_rejects_unbounded_or_secret_shaped_metadata(self) -> None: distributions = [ FakeDistribution( name="govoplan-secret-user@private-host/path", entry_points=[], ), FakeDistribution( version="1.0.0/private-user@host", entry_points=[], ), FakeDistribution( entry_points=[ FakeEntryPoint( name="safe-module", manifest_id="secret-user@private-host/path", ) ], ), ] evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), distributions=distributions, ) encoded = json.dumps(evidence, sort_keys=True) self.assertNotIn("secret-user", encoded) self.assertNotIn("private-host", encoded) self.assertNotIn("private-user", encoded) self.assertTrue( any( item["code"] in { "distribution-metadata-invalid", "module-entry-point-invalid", } for item in evidence["collection_issues"] ) ) def test_pep610_git_provenance_requires_one_coherent_git_form(self) -> None: valid_commit = "a" * 40 cases = ( ( "valid", { "url": "https://example.invalid/govoplan-many.git", "vcs_info": {"vcs": "git", "commit_id": valid_commit}, }, "vcs-commit", ), ( "svn", { "url": "https://example.invalid/govoplan-many.git", "vcs_info": {"vcs": "svn", "commit_id": valid_commit}, }, "malformed-direct-url", ), ( "missing-vcs", { "url": "https://example.invalid/govoplan-many.git", "vcs_info": {"commit_id": valid_commit}, }, "malformed-direct-url", ), ( "ambiguous", { "url": "file:///govoplan-many", "vcs_info": {"vcs": "git", "commit_id": valid_commit}, "dir_info": {"editable": True}, }, "malformed-direct-url", ), ) for label, direct_url, expected_kind in cases: with self.subTest(label=label): evidence = collect_installed_composition( assessment=self.assessment, collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), distributions=[ FakeDistribution(entry_points=[], direct_url=direct_url) ], ) self.assertEqual( expected_kind, evidence["artifacts"][0]["source_provenance"]["kind"], ) 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_boundary_claim_waits_for_anchored_installed_release_origin( 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("blocked", report["status"]) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertIsNone(report["proof_scope"]["target_environment"]["valid"]) self.assertFalse(report["proof_scope"]["external_providers"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) self.assertIn( "boundary_installed_release_origin_untrusted", {item["code"] for item in report["findings"]}, ) self.assertEqual( self.assessment["deployment_profile"]["id"], report["proof_scope"]["target_environment"]["expected_subject_id"], ) self.assertEqual( self.assessment["deployment_profile"]["id"], report["proof_scope"]["target_environment"]["observed_subject_id"], ) def test_external_provider_claim_requires_explicit_matching_subject(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("external_providers", "passed")], allowed_scopes=["external_providers"], ) unconfigured = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) accepted = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, expected_external_provider_subject="provider:test", ) self.assertEqual("blocked", unconfigured["status"]) self.assertIn( "boundary_provider_subject_unconfigured", {item["code"] for item in unconfigured["findings"]}, ) self.assertEqual("blocked", accepted["status"]) provider_scope = accepted["proof_scope"]["external_providers"] self.assertFalse(provider_scope["checked"]) self.assertEqual("provider:test", provider_scope["expected_subject_id"]) self.assertEqual("provider:test", provider_scope["observed_subject_id"]) self.assertIn( "boundary_installed_release_origin_untrusted", {item["code"] for item in accepted["findings"]}, ) def test_boundary_subject_mismatch_blocks_without_echoing_invalid_config( 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", subject_id="another-deployment", ) ], allowed_scopes=["target_environment"], ) mismatch = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) invalid_config = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, expected_external_provider_subject="https://private.invalid/secret", ) self.assertIn( "boundary_claim_subject_mismatch", {item["code"] for item in mismatch["findings"]}, ) encoded = json.dumps(invalid_config, sort_keys=True) self.assertIn("boundary_provider_subject_invalid", encoded) self.assertNotIn("private.invalid", encoded) self.assertNotIn("secret", encoded) def test_authorized_negative_claim_waits_for_installed_origin_binding( 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", "failed"), boundary_claim("production_approval", "rejected"), ], allowed_scopes=["target_environment", "production_approval"], ) 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"]["target_environment"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) self.assertNotIn( "boundary.target_environment.production-like-dev", {item["id"] for item in report["review_targets"]}, ) def test_proof_authorities_must_be_independent_and_key_ids_unambiguous( 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"], ) release_public_key = keyring["keys"][0]["public_key"] reused = deepcopy(authority) reused["keys"][0]["public_key"] = release_public_key duplicate = deepcopy(authority) duplicate["keys"].append( { **deepcopy(duplicate["keys"][0]), "key_id": "authority:duplicate-material", "allowed_scopes": ["production_approval"], } ) reused_report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=reused, ) duplicate_report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=duplicate, ) self.assertIn( "boundary_authority_reuses_release_key", {item["code"] for item in reused_report["findings"]}, ) self.assertIn( "boundary_authority_public_key_duplicate", {item["code"] for item in duplicate_report["findings"]}, ) self.assertFalse( duplicate_report["proof_scope"]["target_environment"]["checked"] ) def test_authority_not_after_is_exclusive(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"], ) authority["keys"][0]["not_after"] = "2026-07-23T12:00:00Z" report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=authority, ) self.assertIn( "boundary_authority_untrusted", {item["code"] for item in report["findings"]}, ) def test_malformed_authority_members_invalidate_the_whole_keyring(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"], ) malformed = deepcopy(authority) malformed["keys"].append( { "key_id": "authority:malformed", "status": "active", "public_key": "not-base64", "allowed_scopes": ["target_environment"], } ) inverted = deepcopy(authority) inverted["keys"][0]["not_before"] = "2026-08-01T00:00:00Z" inverted["keys"][0]["not_after"] = "2026-07-01T00:00:00Z" cases = ( (malformed, "boundary_authority_key_invalid"), (inverted, "boundary_authority_key_interval_invalid"), ) for supplied_keyring, expected_code in cases: with self.subTest(expected_code=expected_code): report = self.review( catalog=catalog, keyring=keyring, installed_evidence=installed, boundary_evidence=proof, authority_keyring=supplied_keyring, ) self.assertEqual("blocked", report["status"]) self.assertIn( expected_code, {item["code"] for item in report["findings"]}, ) self.assertIn( "boundary_authority_untrusted", {item["code"] for item in report["findings"]}, ) self.assertFalse( report["proof_scope"]["target_environment"]["checked"] ) def test_historical_verification_and_generated_runtime_gap_are_visible( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) evidence = matching_installed_evidence(self.assessment) evidence["artifacts"][0]["record_integrity"][ "generated_unhashed_file_count" ] = 3 report = self.review( catalog=catalog, keyring=keyring, installed_evidence=evidence, ) observation = report["proof_scope"]["installed_evidence_observation"] self.assertEqual("historical_override", observation["verification_mode"]) self.assertEqual( 3, report["proof_scope"]["installed_record_integrity"][ "generated_unhashed_file_count" ], ) self.assertIn("HISTORICAL override", render_review(report)) def test_invalid_internal_verification_mode_is_bounded_and_blocking(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, observation_mode="direct_local", verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), verification_mode="https://private.invalid/secret", catalog_trusted=True, ) encoded = json.dumps(result.proof_scope, sort_keys=True) self.assertIn( "installed_evidence_verification_mode", {item.code for item in result.findings}, ) self.assertEqual( "invalid", result.proof_scope["installed_evidence_observation"]["verification_mode"], ) self.assertFalse(result.proof_scope["installed_artifacts"]["checked"]) self.assertNotIn("private.invalid", encoded) 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, installed_evidence_mode: str = "direct_local", installed_schema: dict[str, object] | None = None, expected_external_provider_subject: str | None = None, verification_time: datetime | None = datetime(2026, 7, 23, 12, tzinfo=UTC), ) -> 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=(installed_schema or 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=verification_time, installed_evidence_mode=installed_evidence_mode, expected_external_provider_subject=expected_external_provider_subject, ) def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]: return { "$schema": "./installed-composition-evidence.schema.json", "schema_version": "0.3.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], "collected_at": "2026-07-23T11:58: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": { "basis": "local-pep610-metadata", "kind": "vcs-commit", "commit": commit, }, "record_integrity": { "status": "verified", "hashed_file_count": 1, "permitted_unhashed_file_count": 1, "generated_unhashed_file_count": 0, "unverifiable_file_count": 0, "missing_file_count": 0, "mismatched_file_count": 0, }, } def boundary_claim( scope: str, result: str, *, subject_id: str | None = None, ) -> dict[str, object]: default_subject = ( "provider:test" if scope == "external_providers" else "production-like-dev" ) return { "scope": scope, "result": result, "subject_id": subject_id or default_subject, "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 def create_traversal_distribution( prefix: Path, site_packages: Path, ) -> metadata.PathDistribution: dist_info = site_packages / "govoplan_traversal-1.0.0.dist-info" dist_info.mkdir() metadata_file = dist_info / "METADATA" metadata_file.write_text( "Metadata-Version: 2.1\nName: govoplan-traversal\nVersion: 1.0.0\n", encoding="utf-8", ) unowned_file = prefix / "unowned_runtime" / "private.py" unowned_file.parent.mkdir() unowned_file.write_text("private = True\n", encoding="utf-8") undeclared_script = prefix / "bin" / "govoplan-not-declared" undeclared_script.parent.mkdir() undeclared_script.write_text("#!/bin/sh\n", encoding="utf-8") metadata_path = metadata_file.relative_to(site_packages).as_posix() unowned_path = "../../../unowned_runtime/private.py" record_path = (dist_info / "RECORD").relative_to(site_packages).as_posix() (dist_info / "RECORD").write_text( "\n".join( ( record_row(metadata_path, metadata_file.read_bytes()), record_row(unowned_path, unowned_file.read_bytes()), record_row( "../../../bin/govoplan-not-declared", undeclared_script.read_bytes(), ), f"{record_path},,", ) ) + "\n", encoding="utf-8", ) return metadata.PathDistribution(dist_info) def record_row(path: str, content: bytes) -> str: digest = base64.urlsafe_b64encode(hashlib.sha256(content).digest()).decode("ascii") return f"{path},sha256={digest.rstrip('=')},{len(content)}" class FakeEntryPoint: group = "govoplan.modules" value = "fixture:get_manifest" def __init__( self, *, name: str, manifest_id: str | None = None, manifest_version: str = "1.0.0", ) -> None: self.name = name self.manifest_id = manifest_id or name self.manifest_version = manifest_version def load(self): return lambda: SimpleNamespace( id=self.manifest_id, version=self.manifest_version, ) class FakeDistribution: files = () def __init__( self, *, entry_points: list[FakeEntryPoint], direct_url: dict[str, object] | None = None, name: str = "govoplan-many", version: str = "1.0.0", ) -> None: self.metadata = {"Name": name} self.version = version 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()