from __future__ import annotations import base64 from copy import deepcopy from datetime import UTC, datetime, timedelta import hashlib import importlib.util import json import os from pathlib import Path import subprocess import sys import tempfile from unittest import mock import unittest 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 local_tag_provenance, render_review, review_capability_fit, trusted_keys_from_keyring, ) class CapabilityFitReviewTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.assessment = json.loads( (META_ROOT / "docs" / "capability-fit-current.json").read_text( encoding="utf-8" ) ) cls.schema = json.loads( (META_ROOT / "docs" / "capability-fit.schema.json").read_text( encoding="utf-8" ) ) def test_matching_signed_catalog_is_current(self) -> None: catalog, keyring = signed_catalog(self.assessment) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("current", report["status"]) self.assertEqual([], report["changes"]) self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"]) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertFalse(report["proof_scope"]["local_tag_provenance"]["checked"]) repeated = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual( json.dumps(report, sort_keys=True), json.dumps(repeated, sort_keys=True) ) def test_release_drift_identifies_affected_conclusions(self) -> None: catalog, keyring = signed_catalog( self.assessment, versions={"campaigns": "0.1.11"}, sequence=202607230001 ) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("review_required", report["status"]) target_ids = {item["id"] for item in report["review_targets"]} self.assertIn("campaign.journey", target_ids) self.assertIn("composition.campaigns", target_ids) self.assertIn("assessment.release", target_ids) self.assertIn( "composition_version_changed", {item["code"] for item in report["findings"]} ) def test_contradictory_signed_selected_unit_metadata_blocks_rerun(self) -> None: core = next( item for item in self.assessment["composition"] if item["module_id"] == "core" ) catalog, keyring = signed_catalog( self.assessment, selected_units=[ { "repo": core["repository"], "version": core["manifest_version"], "tag": "v999.0.0", "commit_sha": "0" * 40, "tag_object_sha": "1" * 40, } ], ) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) codes = {item["code"] for item in report["findings"]} self.assertIn("catalog_release_metadata", codes) self.assertIn("composition_commit_changed", codes) self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"]) self.assertFalse(report["proof_scope"]["release_metadata"]["valid"]) def test_signed_selected_unit_commit_drift_requires_review(self) -> None: core = next( item for item in self.assessment["composition"] if item["module_id"] == "core" ) catalog, keyring = signed_catalog( self.assessment, selected_units=[ { "repo": core["repository"], "version": core["manifest_version"], "tag": f"v{core['manifest_version']}", "commit_sha": "0" * 40, "tag_object_sha": "1" * 40, } ], ) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("review_required", report["status"]) self.assertIn( "composition_commit_changed", {item["code"] for item in report["findings"]} ) self.assertIn( "composition.core", {item["id"] for item in report["review_targets"]} ) self.assertFalse(report["proof_scope"]["release_metadata"]["valid"]) def test_missing_duplicate_and_malformed_selected_unit_provenance_blocks( self, ) -> None: core = next( item for item in self.assessment["composition"] if item["module_id"] == "core" ) valid_unit = { "repo": core["repository"], "version": core["manifest_version"], "tag": f"v{core['manifest_version']}", "commit_sha": (str(core["commit"]) + "0" * 40)[:40], "tag_object_sha": "1" * 40, } cases = ( ("missing", None, False, "catalog_source_provenance"), ( "duplicate", [valid_unit, deepcopy(valid_unit)], True, "catalog_release_metadata", ), ( "malformed", [{**valid_unit, "commit_sha": "not-a-git-object"}], True, "catalog_source_provenance", ), ) for label, selected_units, include_selected_units, expected_code in cases: with self.subTest(label=label): catalog, keyring = signed_catalog( self.assessment, selected_units=selected_units, include_selected_units=include_selected_units, ) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) self.assertIn( expected_code, {item["code"] for item in report["findings"]} ) self.assertFalse(report["proof_scope"]["release_metadata"]["valid"]) def test_catalog_validation_ignores_and_preserves_installer_replay_state( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "catalog-sequence.json" state_path.write_text( json.dumps( {"channels": {"stable": {"last_sequence": catalog["sequence"]}}} ), encoding="utf-8", ) replay_environment = { "GOVOPLAN_MODULE_PACKAGE_CATALOG_SEQUENCE_STATE": str(state_path), "GOVOPLAN_MODULE_PACKAGE_CATALOG_ENFORCE_SEQUENCE": "true", } with mock.patch.dict(os.environ, replay_environment, clear=False): report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual( replay_environment, {key: os.environ[key] for key in replay_environment}, ) self.assertEqual("current", report["status"]) self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"]) def test_self_consistent_substitution_fails_independent_trust_root(self) -> None: _, trusted_keyring = signed_catalog(self.assessment) catalog, published_keyring = signed_catalog(self.assessment) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=published_keyring, trusted_keyring=trusted_keyring, ) self.assertEqual("blocked", report["status"]) codes = {item["code"] for item in report["findings"]} self.assertIn("catalog_trust", codes) self.assertNotIn("catalog_keyring_hash_mismatch", codes) self.assertTrue(report["proof_scope"]["assessment_schema"]["valid"]) self.assertFalse( report["proof_scope"]["catalog_signature_and_trusted_keyring"]["valid"] ) self.assertTrue(report["proof_scope"]["published_keyring_hash"]["valid"]) def test_locally_pinned_rotation_keys_fail_closed(self) -> None: keys = trusted_keys_from_keyring( { "keys": [ {"key_id": "active", "status": "active", "public_key": "a"}, {"key_id": "next", "status": "next", "public_key": "b"}, {"key_id": "retired", "status": "retired", "public_key": "c"}, ] } ) self.assertEqual({"active": "a", "next": "b"}, keys) self.assertEqual( {"legacy": "base64-key"}, trusted_keys_from_keyring({"legacy": "base64-key"}), ) self.assertEqual( {}, trusted_keys_from_keyring( { "keys": [ { "key_id": "typo", "status": "retierd", "public_key": "unsafe", } ] } ), ) self.assertEqual( {}, trusted_keys_from_keyring( { "keys": [ {"key_id": "duplicate", "public_key": "a"}, {"key_id": "duplicate", "public_key": "b"}, ] } ), ) self.assertEqual({}, trusted_keys_from_keyring({"keys": "invalid"})) self.assertEqual( {}, trusted_keys_from_keyring( {"keys": [{"key_id": " ", "public_key": "unsafe"}]} ), ) def test_schema_error_blocks_before_comparison(self) -> None: assessment = deepcopy(self.assessment) assessment.pop("scope") catalog, keyring = signed_catalog(self.assessment) report = review_capability_fit( assessment=assessment, schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) self.assertEqual( {"assessment_schema"}, {item["code"] for item in report["findings"]} ) self.assertFalse(report["proof_scope"]["assessment_schema"]["valid"]) self.assertFalse( report["proof_scope"]["catalog_signature_and_keyring"]["checked"] ) self.assertIsNone( report["proof_scope"]["catalog_signature_and_keyring"]["valid"] ) def test_human_report_states_proof_limit(self) -> None: catalog, keyring = signed_catalog(self.assessment) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, ) rendered = render_review(report) self.assertIn("Capability fit rerun: current", rendered) self.assertIn( "No installed-composition, installed-release-origin, target-environment, external-provider, production-approval proof", rendered, ) def test_public_fetch_failure_is_generic_and_blocking(self) -> None: script = META_ROOT / "tools" / "assessments" / "capability-fit.py" spec = importlib.util.spec_from_file_location("capability_fit_cli", script) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) with tempfile.TemporaryDirectory() as temp_dir: trusted_path = Path(temp_dir) / "trusted-keyring.json" trusted_path.write_text('{"keys": []}\n', encoding="utf-8") with mock.patch.object( module, "fetch_json", return_value={ "ok": False, "error": "Authorization: secret at https://internal.invalid", }, ): with self.assertRaisesRegex( SystemExit, r"^Could not fetch fixed public catalog endpoint[.]$" ): module.main( [ "--public", "--trusted-keyring", str(trusted_path), "--skip-tag-provenance", ] ) def test_cli_requires_an_independent_local_trust_root(self) -> None: script = META_ROOT / "tools" / "assessments" / "capability-fit.py" spec = importlib.util.spec_from_file_location( "capability_fit_cli_trust", script ) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) with mock.patch.dict( os.environ, {module.TRUSTED_KEYRING_FILE_ENV: ""}, clear=False, ): with self.assertRaisesRegex(SystemExit, r"--trusted-keyring .* required"): module.main(["--public", "--skip-tag-provenance"]) def test_local_selected_tag_commit_and_object_must_match_exactly(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) commit_sha, tag_object_sha = create_tagged_repository( workspace=workspace, repository="govoplan-core", version="0.1.0", ) assessment = deepcopy(self.assessment) core = next( deepcopy(item) for item in assessment["composition"] if item["module_id"] == "core" ) core["manifest_version"] = "0.1.0" core["commit"] = commit_sha[:12] assessment["composition"] = [core] selected_unit = { "repo": "govoplan-core", "version": "0.1.0", "tag": "v0.1.0", "commit_sha": commit_sha, "tag_object_sha": tag_object_sha, } catalog, keyring = signed_catalog( assessment, selected_units=[selected_unit], ) baseline = review_capability_fit( assessment=deepcopy(assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, workspace_root=workspace, ) self.assertEqual("current", baseline["status"]) self.assertTrue(baseline["proof_scope"]["local_tag_provenance"]["valid"]) forged_commit = commit_sha[:12] + ("0" * 28) if forged_commit == commit_sha: forged_commit = commit_sha[:12] + ("f" * 28) forged_catalog, forged_keyring = signed_catalog( assessment, selected_units=[{**selected_unit, "commit_sha": forged_commit}], ) commit_mismatch = review_capability_fit( assessment=deepcopy(assessment), schema=self.schema, catalog=forged_catalog, published_keyring=forged_keyring, trusted_keyring=forged_keyring, workspace_root=workspace, ) self.assertEqual("review_required", commit_mismatch["status"]) self.assertIn( "tag_signed_commit_changed", {item["kind"] for item in commit_mismatch["changes"]}, ) repo = workspace / "govoplan-core" git_text(repo, "tag", "-d", "v0.1.0") git_text(repo, "tag", "-a", "v0.1.0", "-m", "Re-annotated release") reannotated = review_capability_fit( assessment=deepcopy(assessment), schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, workspace_root=workspace, ) self.assertEqual("review_required", reannotated["status"]) self.assertIn( "tag_object_changed", {item["kind"] for item in reannotated["changes"]}, ) self.assertTrue(reannotated["proof_scope"]["local_tag_provenance"]["checked"]) self.assertFalse(reannotated["proof_scope"]["local_tag_provenance"]["valid"]) def test_missing_composition_entry_does_not_claim_complete_tag_proof(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) core_commit, core_tag_object = create_tagged_repository( workspace=workspace, repository="govoplan-core", version="0.1.0", ) tenancy_commit, _ = create_tagged_repository( workspace=workspace, repository="govoplan-tenancy", version="0.1.0", ) reviewed_assessment = deepcopy(self.assessment) core = next( deepcopy(item) for item in reviewed_assessment["composition"] if item["module_id"] == "core" ) tenancy = next( deepcopy(item) for item in reviewed_assessment["composition"] if item["module_id"] == "tenancy" ) core.update(manifest_version="0.1.0", commit=core_commit[:12]) tenancy.update(manifest_version="0.1.0", commit=tenancy_commit[:12]) catalog_assessment = deepcopy(reviewed_assessment) catalog_assessment["composition"] = [core] reviewed_assessment["composition"] = [core, tenancy] catalog, keyring = signed_catalog( catalog_assessment, selected_units=[ { "repo": "govoplan-core", "version": "0.1.0", "tag": "v0.1.0", "commit_sha": core_commit, "tag_object_sha": core_tag_object, } ], ) report = review_capability_fit( assessment=reviewed_assessment, schema=self.schema, catalog=catalog, published_keyring=keyring, trusted_keyring=keyring, workspace_root=workspace, ) proof = report["proof_scope"]["local_tag_provenance"] self.assertEqual("review_required", report["status"]) self.assertFalse(proof["checked"]) self.assertIsNone(proof["valid"]) self.assertEqual(2, proof["attempted_count"]) self.assertEqual(2, proof["expected_count"]) self.assertNotIn( "tag_provenance_changed", {item["code"] for item in report["findings"]} ) def test_repository_path_traversal_is_not_resolved(self) -> None: with mock.patch("govoplan_assessment.capability_fit.subprocess.run") as run: result = local_tag_provenance( workspace_root=META_ROOT.parent, repository="../outside", version="0.1.0", assessed_commit="0123456", ) self.assertEqual("tag_provenance_unavailable", result["kind"]) self.assertIn("invalid repository identifier", result["message"]) run.assert_not_called() def create_tagged_repository( *, workspace: Path, repository: str, version: str ) -> tuple[str, str]: repo = workspace / repository git_text(workspace, "init", "-b", "main", str(repo)) git_text(repo, "config", "user.name", "Capability Fit Test") git_text(repo, "config", "user.email", "capability-fit@example.invalid") (repo / "README.md").write_text(f"{repository}\n", encoding="utf-8") git_text(repo, "add", "README.md") git_text(repo, "commit", "-m", "Initial release fixture") git_text(repo, "tag", "-a", f"v{version}", "-m", f"Release v{version}") return ( git_text(repo, "rev-parse", "HEAD"), git_text(repo, "rev-parse", f"refs/tags/v{version}"), ) def git_text(cwd: Path, *args: str) -> str: result = subprocess.run( ("git", *args), cwd=cwd, check=False, capture_output=True, text=True, ) if result.returncode != 0: raise AssertionError(result.stderr or result.stdout) return result.stdout.strip() def signed_catalog( assessment: dict[str, object], *, versions: dict[str, str] | None = None, sequence: int = 202607220843, selected_units: list[dict[str, object]] | None = None, include_selected_units: bool = True, ) -> tuple[dict[str, object], dict[str, object]]: versions = versions or {} private_key = Ed25519PrivateKey.generate() public_key = base64.b64encode( private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) ).decode("ascii") keyring: dict[str, object] = { "keyring_version": "1", "purpose": "test", "generated_at": "2026-01-01T00:00:00Z", "keys": [{"key_id": "test-key", "status": "active", "public_key": public_key}], } now = datetime.now(tz=UTC) modules: list[dict[str, object]] = [] core_release: dict[str, object] | None = None for component in assessment["composition"]: module_id = str(component["module_id"]) repository = str(component["repository"]) version = versions.get(module_id, str(component["manifest_version"])) entry: dict[str, object] = { "module_id": module_id, "name": module_id, "version": version, "python_package": repository, "python_ref": f"{repository} @ git+ssh://git@example.invalid/example/{repository}.git@v{version}", } if module_id == "core": core_release = entry else: modules.append(entry) assert core_release is not None if selected_units is None: selected_units = [ { "repo": component["repository"], "version": versions.get( str(component["module_id"]), str(component["manifest_version"]) ), "tag": "v" + versions.get( str(component["module_id"]), str(component["manifest_version"]) ), "commit_sha": (str(component["commit"]) + "0" * 40)[:40], "tag_object_sha": hashlib.sha256( f"{component['repository']}:{component['manifest_version']}:tag".encode() ).hexdigest()[:40], } for component in assessment["composition"] ] release: dict[str, object] = {"keyring_sha256": canonical_hash(keyring)} if include_selected_units: release["selected_units"] = selected_units catalog: dict[str, object] = { "catalog_version": "1", "channel": "stable", "sequence": sequence, "generated_at": now.isoformat().replace("+00:00", "Z"), "expires_at": (now + timedelta(days=30)).isoformat().replace("+00:00", "Z"), "core_release": core_release, "modules": modules, "release": release, } signature_payload = json.dumps( catalog, sort_keys=True, separators=(",", ":"), ensure_ascii=False ).encode("utf-8") catalog["signatures"] = [ { "algorithm": "ed25519", "key_id": "test-key", "value": base64.b64encode(private_key.sign(signature_payload)).decode( "ascii" ), } ] return catalog, keyring def canonical_hash(payload: object) -> str: encoded = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() if __name__ == "__main__": unittest.main()