diff --git a/docs/capability-fit-boundary-evidence.schema.json b/docs/capability-fit-boundary-evidence.schema.json new file mode 100644 index 0000000..948475e --- /dev/null +++ b/docs/capability-fit-boundary-evidence.schema.json @@ -0,0 +1,148 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.add-ideas.de/add-ideas/govoplan/src/branch/main/docs/capability-fit-boundary-evidence.schema.json", + "title": "GovOPlaN externally issued capability-fit boundary evidence", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "evidence_kind", + "proof_id", + "assessment_id", + "assessment_release", + "installed_evidence_sha256", + "issued_at", + "expires_at", + "claims", + "signatures" + ], + "properties": { + "$schema": { + "type": "string", + "format": "uri-reference" + }, + "schema_version": { + "const": "0.1.0" + }, + "evidence_kind": { + "const": "govoplan.capability-fit-boundary-proof" + }, + "proof_id": { + "$ref": "#/$defs/opaque_id" + }, + "assessment_id": { + "$ref": "#/$defs/opaque_id" + }, + "assessment_release": { + "$ref": "#/$defs/opaque_id" + }, + "installed_evidence_sha256": { + "$ref": "#/$defs/sha256" + }, + "issued_at": { + "type": "string", + "format": "date-time" + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "claims": { + "type": "array", + "minItems": 1, + "maxItems": 3, + "items": { + "$ref": "#/$defs/claim" + } + }, + "signatures": { + "type": "array", + "minItems": 1, + "maxItems": 16, + "items": { + "$ref": "#/$defs/signature" + } + } + }, + "$defs": { + "opaque_id": { + "type": "string", + "minLength": 1, + "maxLength": 160, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "claim": { + "type": "object", + "additionalProperties": false, + "required": ["scope", "result", "subject_id", "control_ids", "artifacts"], + "properties": { + "scope": { + "enum": [ + "target_environment", + "external_providers", + "production_approval" + ] + }, + "result": { + "enum": ["passed", "failed", "approved", "rejected"] + }, + "subject_id": { + "$ref": "#/$defs/opaque_id" + }, + "control_ids": { + "type": "array", + "minItems": 1, + "maxItems": 256, + "uniqueItems": true, + "items": { + "$ref": "#/$defs/opaque_id" + } + }, + "artifacts": { + "type": "array", + "minItems": 1, + "maxItems": 256, + "items": { + "$ref": "#/$defs/evidence_artifact" + } + } + } + }, + "evidence_artifact": { + "type": "object", + "additionalProperties": false, + "required": ["artifact_id", "sha256"], + "properties": { + "artifact_id": { + "$ref": "#/$defs/opaque_id" + }, + "sha256": { + "$ref": "#/$defs/sha256" + } + } + }, + "signature": { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "key_id", "value"], + "properties": { + "algorithm": { + "const": "ed25519" + }, + "key_id": { + "$ref": "#/$defs/opaque_id" + }, + "value": { + "type": "string", + "minLength": 1, + "maxLength": 256, + "contentEncoding": "base64" + } + } + }, + "sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + } +} diff --git a/docs/capability-fit-proof-authority-keyring.schema.json b/docs/capability-fit-proof-authority-keyring.schema.json new file mode 100644 index 0000000..4df0bfd --- /dev/null +++ b/docs/capability-fit-proof-authority-keyring.schema.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.add-ideas.de/add-ideas/govoplan/src/branch/main/docs/capability-fit-proof-authority-keyring.schema.json", + "title": "GovOPlaN capability-fit proof authority keyring", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "purpose", "keys"], + "properties": { + "$schema": { + "type": "string", + "format": "uri-reference" + }, + "schema_version": { + "const": "0.1.0" + }, + "purpose": { + "const": "govoplan.capability-fit-proof-authorities" + }, + "keys": { + "type": "array", + "maxItems": 64, + "items": { + "$ref": "#/$defs/key" + } + } + }, + "$defs": { + "opaque_id": { + "type": "string", + "minLength": 1, + "maxLength": 160, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "key": { + "type": "object", + "additionalProperties": false, + "required": ["key_id", "status", "public_key", "allowed_scopes"], + "properties": { + "key_id": { + "$ref": "#/$defs/opaque_id" + }, + "status": { + "enum": ["active", "next", "revoked", "disabled", "retired"] + }, + "public_key": { + "type": "string", + "minLength": 1, + "maxLength": 256, + "contentEncoding": "base64" + }, + "allowed_scopes": { + "type": "array", + "minItems": 1, + "maxItems": 3, + "uniqueItems": true, + "items": { + "enum": [ + "target_environment", + "external_providers", + "production_approval" + ] + } + }, + "not_before": { + "type": "string", + "format": "date-time" + }, + "not_after": { + "type": "string", + "format": "date-time" + } + } + } + } +} diff --git a/docs/installed-composition-evidence.schema.json b/docs/installed-composition-evidence.schema.json new file mode 100644 index 0000000..702e7bd --- /dev/null +++ b/docs/installed-composition-evidence.schema.json @@ -0,0 +1,234 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.add-ideas.de/add-ideas/govoplan/src/branch/main/docs/installed-composition-evidence.schema.json", + "title": "GovOPlaN installed composition evidence", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "evidence_kind", + "assessment_id", + "assessment_release", + "collected_at", + "scope", + "artifacts", + "collection_issues" + ], + "properties": { + "$schema": { + "type": "string", + "format": "uri-reference" + }, + "schema_version": { + "const": "0.1.0" + }, + "evidence_kind": { + "const": "govoplan.installed-composition" + }, + "assessment_id": { + "$ref": "#/$defs/opaque_id" + }, + "assessment_release": { + "$ref": "#/$defs/opaque_id" + }, + "collected_at": { + "type": "string", + "format": "date-time" + }, + "scope": { + "const": "current-python-environment.govoplan-distributions" + }, + "artifacts": { + "type": "array", + "maxItems": 256, + "items": { + "$ref": "#/$defs/artifact" + } + }, + "collection_issues": { + "type": "array", + "maxItems": 256, + "items": { + "$ref": "#/$defs/collection_issue" + } + } + }, + "$defs": { + "opaque_id": { + "type": "string", + "minLength": 1, + "maxLength": 160, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "package_name": { + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$" + }, + "version": { + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._+!-]*$" + }, + "artifact": { + "type": "object", + "additionalProperties": false, + "required": [ + "package_name", + "package_version", + "modules", + "source_provenance", + "record_integrity" + ], + "properties": { + "package_name": { + "$ref": "#/$defs/package_name" + }, + "package_version": { + "$ref": "#/$defs/version" + }, + "modules": { + "type": "array", + "maxItems": 16, + "items": { + "$ref": "#/$defs/module" + } + }, + "source_provenance": { + "$ref": "#/$defs/source_provenance" + }, + "record_integrity": { + "$ref": "#/$defs/record_integrity" + } + } + }, + "module": { + "type": "object", + "additionalProperties": false, + "required": ["module_id", "manifest_version"], + "properties": { + "module_id": { + "$ref": "#/$defs/opaque_id" + }, + "manifest_version": { + "$ref": "#/$defs/version" + } + } + }, + "source_provenance": { + "oneOf": [ + { + "type": "object", + "additionalProperties": false, + "required": ["kind", "commit"], + "properties": { + "kind": { + "enum": ["vcs-commit", "editable-vcs"] + }, + "commit": { + "type": "string", + "pattern": "^[0-9a-f]{40,64}$" + } + } + }, + { + "type": "object", + "additionalProperties": false, + "required": ["kind", "sha256"], + "properties": { + "kind": { + "const": "archive" + }, + "sha256": { + "$ref": "#/$defs/sha256" + } + } + }, + { + "type": "object", + "additionalProperties": false, + "required": ["kind"], + "properties": { + "kind": { + "enum": [ + "editable-local", + "local-directory", + "index-or-unknown", + "malformed-direct-url" + ] + } + } + } + ] + }, + "record_integrity": { + "type": "object", + "additionalProperties": false, + "required": [ + "status", + "hashed_file_count", + "permitted_unhashed_file_count", + "unverifiable_file_count", + "missing_file_count", + "mismatched_file_count" + ], + "properties": { + "status": { + "enum": [ + "verified", + "partial", + "mismatch", + "unavailable", + "limit-exceeded" + ] + }, + "hashed_file_count": { + "$ref": "#/$defs/count" + }, + "permitted_unhashed_file_count": { + "$ref": "#/$defs/count" + }, + "unverifiable_file_count": { + "$ref": "#/$defs/count" + }, + "missing_file_count": { + "$ref": "#/$defs/count" + }, + "mismatched_file_count": { + "$ref": "#/$defs/count" + } + } + }, + "collection_issue": { + "type": "object", + "additionalProperties": false, + "required": ["code", "package_name"], + "properties": { + "code": { + "enum": [ + "distribution-metadata-invalid", + "duplicate-distribution", + "module-entry-point-load-failed", + "module-entry-point-invalid", + "module-entry-point-limit-exceeded", + "collection-limit-exceeded" + ] + }, + "package_name": { + "$ref": "#/$defs/package_name" + } + } + }, + "count": { + "type": "integer", + "minimum": 0, + "maximum": 1000000 + }, + "sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + } +} diff --git a/tests/test_capability_fit_evidence.py b/tests/test_capability_fit_evidence.py new file mode 100644 index 0000000..0b12378 --- /dev/null +++ b/tests/test_capability_fit_evidence.py @@ -0,0 +1,617 @@ +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 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) + + 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") + 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() diff --git a/tools/assessments/capability-fit.py b/tools/assessments/capability-fit.py index 645975a..421b5bb 100755 --- a/tools/assessments/capability-fit.py +++ b/tools/assessments/capability-fit.py @@ -17,11 +17,18 @@ 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 import render_review, review_capability_fit # noqa: E402 +from govoplan_assessment import ( # noqa: E402 + collect_installed_composition, + render_review, + review_capability_fit, +) +from govoplan_assessment.evidence import validate_payload # noqa: E402 from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402 TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE" +MAX_ASSESSMENT_INPUT_BYTES = 16 * 1024 * 1024 +MAX_EVIDENCE_INPUT_BYTES = 4 * 1024 * 1024 def parse_args(argv: list[str] | None = None) -> argparse.Namespace: @@ -71,6 +78,58 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: action="store_true", help="Compare signed metadata without checking local annotated tags.", ) + installed = parser.add_mutually_exclusive_group() + installed.add_argument( + "--installed-evidence", + type=Path, + help="Validate an existing bounded installed-composition evidence document.", + ) + installed.add_argument( + "--collect-installed-evidence", + type=Path, + help=( + "Collect the current interpreter's GovOPlaN distributions, write the " + "bounded evidence document, and include it in this review. Loading " + "module entry points executes installed GovOPlaN manifest factories." + ), + ) + parser.add_argument( + "--installed-evidence-schema", + type=Path, + default=META_ROOT / "docs" / "installed-composition-evidence.schema.json", + ) + parser.add_argument( + "--boundary-evidence", + type=Path, + help=( + "Externally issued target/provider/production proof bound to the " + "installed-evidence digest." + ), + ) + parser.add_argument( + "--boundary-evidence-schema", + type=Path, + default=META_ROOT / "docs" / "capability-fit-boundary-evidence.schema.json", + ) + parser.add_argument( + "--boundary-authority-keyring", + type=Path, + help=( + "Independently provisioned keyring authorizing proof signers for " + "specific target/provider/production scopes." + ), + ) + parser.add_argument( + "--boundary-authority-keyring-schema", + type=Path, + default=META_ROOT + / "docs" + / "capability-fit-proof-authority-keyring.schema.json", + ) + parser.add_argument( + "--verification-time", + help="Optional ISO-8601 time for reproducible boundary-proof verification.", + ) parser.add_argument( "--json", action="store_true", help="Print the machine-readable review report." ) @@ -88,6 +147,22 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit("--keyring is required with --catalog") if args.public and args.keyring is not None: raise SystemExit("--keyring cannot be combined with --public") + if (args.boundary_evidence is None) != (args.boundary_authority_keyring is None): + raise SystemExit( + "--boundary-evidence and --boundary-authority-keyring are required together" + ) + if args.boundary_evidence is not None and ( + args.installed_evidence is None and args.collect_installed_evidence is None + ): + raise SystemExit( + "--boundary-evidence requires --installed-evidence or --collect-installed-evidence" + ) + if ( + args.collect_installed_evidence is not None + and args.output is not None + and args.collect_installed_evidence.resolve() == args.output.resolve() + ): + raise SystemExit("Evidence and review output paths must differ") configured_trusted_keyring = os.getenv(TRUSTED_KEYRING_FILE_ENV, "").strip() trusted_keyring_path = args.trusted_keyring or ( Path(configured_trusted_keyring) if configured_trusted_keyring else None @@ -96,8 +171,16 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit( f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required" ) - assessment = read_object(args.assessment, label="assessment") - schema = read_object(args.schema, label="schema") + assessment = read_object( + args.assessment, + label="assessment", + max_bytes=MAX_ASSESSMENT_INPUT_BYTES, + ) + schema = read_object( + args.schema, + label="schema", + max_bytes=MAX_ASSESSMENT_INPUT_BYTES, + ) if args.public: catalog = fetch_public_json( f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/stable.json", @@ -108,13 +191,75 @@ def main(argv: list[str] | None = None) -> int: label="published keyring", ) else: - catalog = read_object(args.catalog, label="catalog") - published_keyring = read_object(args.keyring, label="published keyring") + catalog = read_object( + args.catalog, label="catalog", max_bytes=MAX_ASSESSMENT_INPUT_BYTES + ) + published_keyring = read_object( + args.keyring, + label="published keyring", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) trusted_keyring = read_object( trusted_keyring_path, label="trusted keyring", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) + installed_evidence = None + installed_evidence_schema = None + if ( + args.installed_evidence is not None + or args.collect_installed_evidence is not None + ): + installed_evidence_schema = read_object( + args.installed_evidence_schema, + label="installed evidence schema", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + if args.installed_evidence is not None: + installed_evidence = read_object( + args.installed_evidence, + label="installed evidence", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + else: + installed_evidence = collect_installed_composition(assessment=assessment) + collection_schema_errors = validate_payload( + payload=installed_evidence, + schema=installed_evidence_schema, + ) + if collection_schema_errors: + raise SystemExit( + "Collected installed evidence did not satisfy its bounded schema" + ) + write_object(args.collect_installed_evidence, installed_evidence) + + boundary_evidence = None + boundary_evidence_schema = None + boundary_authority_keyring = None + boundary_authority_keyring_schema = None + if args.boundary_evidence is not None: + boundary_evidence = read_object( + args.boundary_evidence, + label="boundary evidence", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + boundary_evidence_schema = read_object( + args.boundary_evidence_schema, + label="boundary evidence schema", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + boundary_authority_keyring = read_object( + args.boundary_authority_keyring, + label="boundary authority keyring", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + boundary_authority_keyring_schema = read_object( + args.boundary_authority_keyring_schema, + label="boundary authority keyring schema", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + report = review_capability_fit( assessment=assessment, schema=schema, @@ -124,6 +269,13 @@ def main(argv: list[str] | None = None) -> int: workspace_root=None if args.skip_tag_provenance else args.workspace_root.resolve(), + installed_evidence=installed_evidence, + installed_evidence_schema=installed_evidence_schema, + boundary_evidence=boundary_evidence, + boundary_evidence_schema=boundary_evidence_schema, + boundary_authority_keyring=boundary_authority_keyring, + boundary_authority_keyring_schema=boundary_authority_keyring_schema, + verification_time=parse_datetime(args.verification_time), ) encoded = json.dumps(report, indent=2, sort_keys=True) + "\n" if args.output is not None: @@ -133,18 +285,54 @@ def main(argv: list[str] | None = None) -> int: return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]] -def read_object(path: Path | None, *, label: str) -> dict[str, object]: +def read_object(path: Path | None, *, label: str, max_bytes: int) -> dict[str, object]: if path is None: raise SystemExit(f"Missing {label} path") try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: + if path.stat().st_size > max_bytes: + raise SystemExit( + f"{label.capitalize()} exceeds the {max_bytes}-byte input limit" + ) + with path.open("rb") as handle: + encoded = handle.read(max_bytes + 1) + if len(encoded) > max_bytes: + raise SystemExit( + f"{label.capitalize()} exceeds the {max_bytes}-byte input limit" + ) + payload = json.loads(encoded.decode("utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise SystemExit(f"Could not read {label} {path}: {exc}") from exc if not isinstance(payload, dict): raise SystemExit(f"{label.capitalize()} must be a JSON object: {path}") return payload +def write_object(path: Path | None, payload: dict[str, object]) -> None: + if path is None: + raise SystemExit("Missing installed evidence output path") + encoded = json.dumps(payload, indent=2, sort_keys=True) + "\n" + if len(encoded.encode("utf-8")) > MAX_EVIDENCE_INPUT_BYTES: + raise SystemExit( + "Collected installed evidence exceeds the bounded output limit" + ) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(encoded, encoding="utf-8") + + +def parse_datetime(value: str | None): + if value is None: + return None + from datetime import datetime + + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise SystemExit("--verification-time must be an ISO-8601 timestamp") from exc + if parsed.tzinfo is None: + raise SystemExit("--verification-time must include a timezone") + return parsed + + def fetch_public_json(url: str, *, label: str) -> dict[str, object]: result = fetch_json(url) if result.get("ok") is not True or not isinstance(result.get("payload"), dict): diff --git a/tools/assessments/govoplan_assessment/__init__.py b/tools/assessments/govoplan_assessment/__init__.py index 05fdc9e..445bf8e 100644 --- a/tools/assessments/govoplan_assessment/__init__.py +++ b/tools/assessments/govoplan_assessment/__init__.py @@ -1,5 +1,10 @@ """Repeatable GovOPlaN product-assessment tooling.""" from .capability_fit import review_capability_fit, render_review +from .evidence import collect_installed_composition -__all__ = ("render_review", "review_capability_fit") +__all__ = ( + "collect_installed_composition", + "render_review", + "review_capability_fit", +) diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index e624a33..97c621e 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -25,6 +25,13 @@ import govoplan_core.core.module_package_catalog as module_package_catalog from govoplan_release.source_provenance import catalog_source_selection from govoplan_release.version_alignment import candidate_catalog_version_issues +from .evidence import ( + BoundaryEvidenceReview, + InstalledEvidenceReview, + review_boundary_evidence, + review_installed_composition, +) + RELEASE_REF_PATTERN = re.compile( r"^(?P[a-z][a-z0-9_-]*)-catalog-(?P[0-9]+)$" @@ -49,6 +56,13 @@ def review_capability_fit( published_keyring: dict[str, Any], trusted_keyring: dict[str, Any], workspace_root: Path | None = None, + installed_evidence: dict[str, Any] | None = None, + installed_evidence_schema: dict[str, Any] | None = None, + boundary_evidence: dict[str, Any] | None = None, + boundary_evidence_schema: dict[str, Any] | None = None, + boundary_authority_keyring: dict[str, Any] | None = None, + boundary_authority_keyring_schema: dict[str, Any] | None = None, + verification_time: datetime | None = None, ) -> dict[str, Any]: """Return a secret-free release-drift report for one fit assessment.""" @@ -68,6 +82,8 @@ def review_capability_fit( local_tag_provenance_attempted=0, local_tag_provenance_expected=0, local_tag_catalog_scope_complete=False, + installed_review=None, + boundary_review=None, ) catalog_validation = validate_catalog( @@ -340,11 +356,47 @@ def review_capability_fit( ) ) + installed_review = review_installed_composition( + assessment=assessment, + catalog_entries=catalog_entries, + selected_versions=source_selection.selected_versions, + selected_commits=source_selection.selected_commits, + evidence=installed_evidence, + schema=installed_evidence_schema, + ) + findings.extend( + Finding(item.severity, item.code, item.message, item.assessment_ids) + for item in installed_review.findings + ) + changes.extend(installed_review.changes) + changed_repositories.update(installed_review.changed_repositories) + + boundary_review = review_boundary_evidence( + assessment=assessment, + installed_review=installed_review, + evidence=boundary_evidence, + evidence_schema=boundary_evidence_schema, + authority_keyring=boundary_authority_keyring, + authority_keyring_schema=boundary_authority_keyring_schema, + verification_time=verification_time, + ) + findings.extend( + Finding(item.severity, item.code, item.message, item.assessment_ids) + for item in boundary_review.findings + ) + review_targets = conclusions_needing_review( assessment=assessment, changed_repositories=changed_repositories, release_changed=assessed_sequence != catalog_sequence, ) + review_targets = merge_review_targets( + review_targets, + installed_evidence_review_targets( + assessment=assessment, + installed_review=installed_review, + ), + ) return build_report( assessment=assessment, catalog=catalog, @@ -355,6 +407,8 @@ def review_capability_fit( local_tag_provenance_attempted=local_tag_provenance_attempted, local_tag_provenance_expected=local_tag_provenance_expected, local_tag_catalog_scope_complete=local_tag_catalog_scope_complete, + installed_review=installed_review, + boundary_review=boundary_review, ) @@ -512,6 +566,7 @@ def catalog_entries_by_id( entries[module_id] = { "module_id": module_id, "repository": catalog_repository(item), + "package": catalog_package(item), "version": _text(item.get("version")), "tags": tuple( tag for tag in (catalog_ref_tag(value) for value in refs) if tag @@ -534,6 +589,11 @@ def catalog_repository(entry: dict[str, Any]) -> str | None: return None +def catalog_package(entry: dict[str, Any]) -> str | None: + package = _text(entry.get("python_package")) + return package.split("[", 1)[0].strip() if package else None + + def catalog_ref_tag(value: str) -> str | None: match = TAG_PATTERN.search(value) return match.group("tag") if match else None @@ -709,6 +769,51 @@ def conclusions_needing_review( return [targets[key] for key in sorted(targets)] +def installed_evidence_review_targets( + *, + assessment: dict[str, Any], + installed_review: InstalledEvidenceReview, +) -> list[dict[str, Any]]: + if not installed_review.findings: + return [] + targets: dict[str, dict[str, Any]] = { + "assessment.installed_composition": { + "id": "assessment.installed_composition", + "section": "installed_composition", + "reason": "Installed-composition evidence is incomplete, mutable, extra, missing, or inconsistent with the assessed release.", + } + } + components = { + str(item.get("module_id")): item + for item in assessment.get("composition", []) + if isinstance(item, dict) + } + for module_id in sorted(installed_review.affected_module_ids): + component = components.get(module_id) + target_id = f"composition.{module_id}" + targets[target_id] = { + "id": target_id, + "section": "composition", + "repositories": [component.get("repository")] + if component is not None + else [], + "reason": "The observed installed distribution, manifest, RECORD integrity, or immutable provenance differs from the assessed composition.", + } + return [targets[key] for key in sorted(targets)] + + +def merge_review_targets( + *groups: Iterable[dict[str, Any]], +) -> list[dict[str, Any]]: + merged: dict[str, dict[str, Any]] = {} + for group in groups: + for item in group: + item_id = str(item.get("id") or "") + if item_id and item_id not in merged: + merged[item_id] = item + return [merged[key] for key in sorted(merged)] + + def build_report( *, assessment: dict[str, Any], @@ -720,6 +825,8 @@ def build_report( local_tag_provenance_attempted: int, local_tag_provenance_expected: int, local_tag_catalog_scope_complete: bool, + installed_review: InstalledEvidenceReview | None, + boundary_review: BoundaryEvidenceReview | None, ) -> dict[str, Any]: finding_list = sorted( findings, @@ -736,7 +843,9 @@ def build_report( ) catalog_valid = trusted_signature_valid and published_keyring_valid release_valid = catalog_valid and not any( - item.severity in {"blocker", "review"} for item in finding_list + item.severity in {"blocker", "review"} + and not item.code.startswith(("installed_", "boundary_")) + for item in finding_list ) tag_provenance_checked = ( local_tag_provenance_expected > 0 @@ -752,8 +861,54 @@ def build_report( status = "review_required" else: status = "current" + installed_scope = ( + installed_review.proof_scope + if installed_review is not None + else { + "installed_artifacts": {"checked": False, "valid": None}, + "installed_record_integrity": {"checked": False, "valid": None}, + "installed_source_provenance": {"checked": False, "valid": None}, + "runtime_activation": {"checked": False, "valid": None}, + } + ) + boundary_scope = ( + boundary_review.proof_scope + if boundary_review is not None + else { + "target_environment": {"checked": False, "valid": None}, + "external_providers": {"checked": False, "valid": None}, + "production_approval": {"checked": False, "valid": None}, + } + ) + proof_scope = { + "assessment_schema": {"checked": True, "valid": schema_valid}, + "catalog_signature_and_keyring": { + "checked": catalog_checked, + "valid": catalog_valid if catalog_checked else None, + }, + "catalog_signature_and_trusted_keyring": { + "checked": catalog_checked, + "valid": trusted_signature_valid if catalog_checked else None, + }, + "published_keyring_hash": { + "checked": catalog_checked, + "valid": published_keyring_valid if catalog_checked else None, + }, + "release_metadata": { + "checked": catalog_checked, + "valid": release_valid if catalog_checked else None, + }, + "local_tag_provenance": { + "checked": tag_provenance_checked, + "valid": tag_provenance_valid if tag_provenance_checked else None, + "attempted_count": local_tag_provenance_attempted, + "expected_count": local_tag_provenance_expected, + }, + **installed_scope, + **boundary_scope, + } return { - "report_version": "0.1.0", + "report_version": "0.2.0", "status": status, "assessment_id": assessment.get("assessment_id"), "assessment_release": assessment.get("release", {}).get("ref") @@ -764,35 +919,7 @@ def build_report( "sequence": catalog.get("sequence"), "generated_at": catalog.get("generated_at"), }, - "proof_scope": { - "assessment_schema": {"checked": True, "valid": schema_valid}, - "catalog_signature_and_keyring": { - "checked": catalog_checked, - "valid": catalog_valid if catalog_checked else None, - }, - "catalog_signature_and_trusted_keyring": { - "checked": catalog_checked, - "valid": trusted_signature_valid if catalog_checked else None, - }, - "published_keyring_hash": { - "checked": catalog_checked, - "valid": published_keyring_valid if catalog_checked else None, - }, - "release_metadata": { - "checked": catalog_checked, - "valid": release_valid if catalog_checked else None, - }, - "local_tag_provenance": { - "checked": tag_provenance_checked, - "valid": tag_provenance_valid if tag_provenance_checked else None, - "attempted_count": local_tag_provenance_attempted, - "expected_count": local_tag_provenance_expected, - }, - "installed_artifacts": {"checked": False, "valid": None}, - "target_environment": {"checked": False, "valid": None}, - "external_providers": {"checked": False, "valid": None}, - "production_approval": {"checked": False, "valid": None}, - }, + "proof_scope": proof_scope, "findings": [asdict(item) for item in finding_list], "changes": sorted( changes, @@ -803,11 +930,22 @@ def build_report( def render_review(report: dict[str, Any]) -> str: + installed_checked = ( + report.get("proof_scope", {}).get("installed_artifacts", {}).get("checked") + is True + ) lines = [ f"Capability fit rerun: {report['status']}", f"Assessment: {report.get('assessment_id') or '-'} ({report.get('assessment_release') or '-'})", f"Catalog: {report['catalog'].get('channel') or '-'} sequence {report['catalog'].get('sequence') or '-'}", - "Scope: schema, signed release metadata, and optional local tag provenance only.", + ( + "Scope: schema, signed release metadata, optional local tag provenance, " + + ( + "and supplied installed-composition evidence." + if installed_checked + else "and no installed-composition evidence." + ) + ), ] findings = report.get("findings") or [] if findings: @@ -822,9 +960,23 @@ def render_review(report: dict[str, Any]) -> str: if targets: lines.append("Conclusions requiring review:") lines.extend(f"- {item['id']}: {item['reason']}" for item in targets) - lines.append( - "No installed-artifact, target-provider, or production proof was performed." - ) + unchecked_boundaries = [ + label + for key, label in ( + ("target_environment", "target-environment"), + ("external_providers", "external-provider"), + ("production_approval", "production-approval"), + ) + if report.get("proof_scope", {}).get(key, {}).get("checked") is not True + ] + if not installed_checked: + unchecked_boundaries.insert(0, "installed-artifact") + if not installed_checked and len(unchecked_boundaries) == 4: + lines.append( + "No installed-artifact, target-provider, or production proof was performed." + ) + elif unchecked_boundaries: + lines.append("No " + ", ".join(unchecked_boundaries) + " proof was performed.") return "\n".join(lines) + "\n" diff --git a/tools/assessments/govoplan_assessment/evidence.py b/tools/assessments/govoplan_assessment/evidence.py new file mode 100644 index 0000000..0fe0ade --- /dev/null +++ b/tools/assessments/govoplan_assessment/evidence.py @@ -0,0 +1,1322 @@ +"""Bounded installed-composition and externally authorized fit evidence.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from datetime import UTC, datetime +import hashlib +import hmac +from importlib import metadata +import json +from pathlib import Path +import re +import sys +from typing import Any, Iterable, Mapping, Sequence + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from jsonschema import Draft202012Validator, FormatChecker +from jsonschema.exceptions import SchemaError + + +MAX_GOVOPLAN_DISTRIBUTIONS = 256 +MAX_DISTRIBUTION_FILES = 10_000 +MAX_COLLECTION_RECORD_FILES = 50_000 +MAX_HASHED_FILE_BYTES = 64 * 1024 * 1024 +MAX_HASHED_DISTRIBUTION_BYTES = 512 * 1024 * 1024 +MAX_HASHED_COLLECTION_BYTES = 2 * 1024 * 1024 * 1024 +MAX_DIRECT_URL_BYTES = 64 * 1024 +PACKAGE_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +OPAQUE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +VERSION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]*$") +GIT_OBJECT_PATTERN = re.compile(r"^[0-9a-f]{40,64}$") +SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") +POSITIVE_RESULTS = { + "target_environment": "passed", + "external_providers": "passed", + "production_approval": "approved", +} +NEGATIVE_RESULTS = { + "target_environment": "failed", + "external_providers": "failed", + "production_approval": "rejected", +} +PROOF_REQUIREMENTS = { + "target_environment": ( + "A target-run result bound to this assessment release and installed-evidence digest.", + "An independently trusted authority key permitted for target_environment proof.", + "Opaque control IDs and content hashes for the target acceptance artifacts.", + ), + "external_providers": ( + "Provider acceptance results bound to this assessment release and installed-evidence digest.", + "An independently trusted authority key permitted for external_providers proof.", + "Opaque control IDs and content hashes; credentials and endpoint identifiers stay outside the report.", + ), + "production_approval": ( + "An explicit approval bound to this assessment release and installed-evidence digest.", + "An independently provisioned authority key permitted for production_approval.", + "An unexpired signed proof bundle; an assessment or operator cannot approve itself.", + ), +} + + +@dataclass(frozen=True, slots=True) +class EvidenceFinding: + severity: str + code: str + message: str + assessment_ids: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class InstalledEvidenceReview: + findings: tuple[EvidenceFinding, ...] + changes: tuple[dict[str, Any], ...] + changed_repositories: frozenset[str] + affected_module_ids: frozenset[str] + proof_scope: Mapping[str, Any] + evidence_sha256: str | None + + +@dataclass(frozen=True, slots=True) +class BoundaryEvidenceReview: + findings: tuple[EvidenceFinding, ...] + proof_scope: Mapping[str, Any] + + +@dataclass(slots=True) +class CollectionHashBudget: + remaining_files: int = MAX_COLLECTION_RECORD_FILES + remaining_bytes: int = MAX_HASHED_COLLECTION_BYTES + + +def validate_payload( + *, payload: dict[str, Any], schema: dict[str, Any] +) -> tuple[str, ...]: + try: + Draft202012Validator.check_schema(schema) + except SchemaError as exc: + return (f"$schema: {exc.message}",) + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + return tuple( + f"{_json_path(error.absolute_path)}: failed {error.validator!r} validation" + for error in sorted( + validator.iter_errors(payload), + key=lambda item: ( + tuple(str(part) for part in item.absolute_path), + str(item.validator), + item.message, + ), + ) + ) + + +def collect_installed_composition( + *, + assessment: dict[str, Any], + collected_at: datetime | None = None, + distributions: Iterable[metadata.Distribution] | None = None, +) -> dict[str, Any]: + """Collect only bounded, non-identifying GovOPlaN distribution evidence. + + Loading ``govoplan.modules`` entry points executes installed module code. The + collector catches failures and records only a stable error code; it never + serializes exception text, URLs, local paths, hostnames, or usernames. + """ + + artifacts: list[dict[str, Any]] = [] + issues: list[dict[str, str]] = [] + hash_budget = CollectionHashBudget() + candidates: list[tuple[str, metadata.Distribution]] = [] + source = metadata.distributions() if distributions is None else distributions + for distribution in source: + try: + package_name = normalize_package_name(str(distribution.metadata["Name"])) + except (Exception, SystemExit): + continue + if package_name.startswith("govoplan-"): + candidates.append((package_name, distribution)) + + candidates.sort(key=lambda item: (item[0], _distribution_version(item[1]))) + if len(candidates) > MAX_GOVOPLAN_DISTRIBUTIONS: + candidates = candidates[:MAX_GOVOPLAN_DISTRIBUTIONS] + issues.append( + { + "code": "collection-limit-exceeded", + "package_name": "govoplan-collector", + } + ) + + name_counts: dict[str, int] = {} + for package_name, _ in candidates: + name_counts[package_name] = name_counts.get(package_name, 0) + 1 + for package_name, count in sorted(name_counts.items()): + if count > 1: + issues.append( + { + "code": "duplicate-distribution", + "package_name": package_name, + } + ) + + for package_name, distribution in candidates: + version = _distribution_version(distribution) + if not version or PACKAGE_PATTERN.fullmatch(package_name) is None: + issues.append( + { + "code": "distribution-metadata-invalid", + "package_name": package_name + if PACKAGE_PATTERN.fullmatch(package_name) + else "govoplan-invalid", + } + ) + continue + modules, module_issues = _distribution_modules( + package_name=package_name, + distribution=distribution, + ) + issues.extend(module_issues) + artifacts.append( + { + "package_name": package_name, + "package_version": version, + "modules": modules, + "source_provenance": _source_provenance(distribution), + "record_integrity": _record_integrity( + distribution, + budget=hash_budget, + ), + } + ) + + observed_at = collected_at or datetime.now(tz=UTC) + if observed_at.tzinfo is None: + observed_at = observed_at.replace(tzinfo=UTC) + release = assessment.get("release") + assessment_release = release.get("ref") if isinstance(release, dict) else None + sorted_issues = sorted( + _unique_dicts(issues), + key=lambda item: (item["package_name"], item["code"]), + ) + if len(sorted_issues) > 256: + limit_issue = { + "code": "collection-limit-exceeded", + "package_name": "govoplan-collector", + } + sorted_issues = sorted_issues[:255] + if limit_issue not in sorted_issues: + sorted_issues.append(limit_issue) + sorted_issues.sort(key=lambda item: (item["package_name"], item["code"])) + return { + "$schema": "./installed-composition-evidence.schema.json", + "schema_version": "0.1.0", + "evidence_kind": "govoplan.installed-composition", + "assessment_id": str(assessment.get("assessment_id") or "invalid-assessment"), + "assessment_release": str(assessment_release or "invalid-release"), + "collected_at": observed_at.astimezone(UTC).isoformat().replace("+00:00", "Z"), + "scope": "current-python-environment.govoplan-distributions", + "artifacts": sorted( + artifacts, + key=lambda item: ( + item["package_name"], + item["package_version"], + canonical_bytes(item), + ), + ), + "collection_issues": sorted_issues, + } + + +def review_installed_composition( + *, + assessment: dict[str, Any], + catalog_entries: Mapping[str, Mapping[str, Any]], + selected_versions: Mapping[str, str], + selected_commits: Mapping[str, str], + evidence: dict[str, Any] | None, + schema: dict[str, Any] | None, +) -> InstalledEvidenceReview: + if evidence is None: + return InstalledEvidenceReview( + findings=(), + changes=(), + changed_repositories=frozenset(), + affected_module_ids=frozenset(), + proof_scope=_unchecked_installed_scope(), + evidence_sha256=None, + ) + digest = canonical_sha256(evidence) + if schema is None: + finding = EvidenceFinding( + "blocker", + "installed_evidence_schema_missing", + "Installed evidence was supplied without its validation schema.", + ("assessment.installed_composition",), + ) + return InstalledEvidenceReview( + findings=(finding,), + changes=(), + changed_repositories=frozenset(), + affected_module_ids=frozenset(), + proof_scope=_invalid_installed_scope(digest=digest), + evidence_sha256=digest, + ) + schema_errors = validate_payload(payload=evidence, schema=schema) + if schema_errors: + findings = tuple( + EvidenceFinding( + "blocker", + "installed_evidence_schema", + message, + ("assessment.installed_composition",), + ) + for message in schema_errors + ) + return InstalledEvidenceReview( + findings=findings, + changes=(), + changed_repositories=frozenset(), + affected_module_ids=frozenset(), + proof_scope=_invalid_installed_scope(digest=digest), + evidence_sha256=digest, + ) + + findings: list[EvidenceFinding] = [] + changes: list[dict[str, Any]] = [] + changed_repositories: set[str] = set() + affected_module_ids: set[str] = set() + expected_release = ( + assessment.get("release", {}).get("ref") + if isinstance(assessment.get("release"), dict) + else None + ) + binding_valid = True + if evidence.get("assessment_id") != assessment.get("assessment_id"): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_evidence_assessment_mismatch", + "Installed evidence is bound to a different assessment ID.", + ("assessment.installed_composition",), + ) + ) + if evidence.get("assessment_release") != expected_release: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_evidence_release_mismatch", + "Installed evidence is bound to a different assessment release.", + ("assessment.release", "assessment.installed_composition"), + ) + ) + + artifact_rows = [ + item for item in evidence.get("artifacts", []) if isinstance(item, dict) + ] + artifacts_by_package: dict[str, dict[str, Any]] = {} + duplicate_packages: set[str] = set() + for artifact in artifact_rows: + package_name = normalize_package_name(str(artifact.get("package_name") or "")) + if package_name in artifacts_by_package: + duplicate_packages.add(package_name) + else: + artifacts_by_package[package_name] = artifact + for package_name in sorted(duplicate_packages): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_distribution_duplicate", + f"Installed evidence repeats distribution {package_name!r}.", + ("assessment.installed_composition",), + ) + ) + + expected_by_package: dict[str, tuple[dict[str, Any], Mapping[str, Any]]] = {} + assessment_by_module: dict[str, dict[str, Any]] = {} + for component in assessment.get("composition", []): + if not isinstance(component, dict): + continue + module_id = str(component.get("module_id") or "") + assessment_by_module[module_id] = component + if component.get("enabled") is not True: + continue + catalog_entry = catalog_entries.get(module_id) + if catalog_entry is None: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_expected_catalog_entry_missing", + f"Enabled module {module_id!r} cannot be mapped to a catalog package.", + (f"composition.{module_id}",), + ) + ) + continue + package_name = normalize_package_name(str(catalog_entry.get("package") or "")) + if not package_name: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_expected_package_missing", + f"Enabled module {module_id!r} has no valid catalog package name.", + (f"composition.{module_id}",), + ) + ) + continue + if package_name in expected_by_package: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_expected_package_ambiguous", + f"Multiple enabled composition entries map to {package_name!r}.", + ("assessment.installed_composition",), + ) + ) + continue + expected_by_package[package_name] = (component, catalog_entry) + + if not expected_by_package: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installed_expected_composition_empty", + "Installed proof requires at least one enabled assessment component mapped to a catalog package.", + ("assessment.installed_composition",), + ) + ) + + record_expected = 0 + record_verified = 0 + provenance_expected = 0 + provenance_anchored = 0 + provenance_mutable = 0 + provenance_unanchored = 0 + artifact_valid = binding_valid and not duplicate_packages + + for package_name, (component, catalog_entry) in sorted(expected_by_package.items()): + module_id = str(component["module_id"]) + repository = str(component["repository"]) + artifact = artifacts_by_package.get(package_name) + record_expected += 1 + provenance_expected += 1 + if package_name in duplicate_packages: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_distribution_duplicate", + code="installed_distribution_duplicate_ambiguous", + message=f"Installed distribution {package_name!r} is duplicated, so no candidate was selected for integrity or provenance conclusions.", + ) + continue + if artifact is None: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_distribution_missing", + code="installed_distribution_missing", + message=f"Enabled module {module_id!r} has no installed distribution {package_name!r} in the evidence.", + ) + continue + + expected_version = str(component["manifest_version"]) + package_version = str(artifact.get("package_version") or "") + catalog_version = str(catalog_entry.get("version") or "") + if package_version != expected_version or package_version != catalog_version: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_version_mismatch", + code="installed_distribution_version_mismatch", + message=( + f"Installed {package_name!r} version {package_version!r} does not match " + f"assessment {expected_version!r} and catalog {catalog_version!r}." + ), + extra={ + "installed_version": package_version, + "assessed_version": expected_version, + "catalog_version": catalog_version, + }, + ) + + modules = [ + item for item in artifact.get("modules", []) if isinstance(item, dict) + ] + module_rows: dict[str, dict[str, Any]] = {} + duplicate_module_ids: set[str] = set() + for item in modules: + observed_id = str(item.get("module_id") or "") + if observed_id in module_rows: + duplicate_module_ids.add(observed_id) + else: + module_rows[observed_id] = item + expected_manifest = module_rows.get(module_id) + manifest_mismatch = ( + expected_manifest is None if module_id != "core" else False + ) or ( + expected_manifest is not None + and str(expected_manifest.get("manifest_version") or "") != expected_version + ) + if manifest_mismatch: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_manifest_mismatch", + code="installed_manifest_version_mismatch", + message=f"Installed distribution {package_name!r} does not expose module {module_id!r} at version {expected_version!r}.", + ) + unexpected_modules = sorted(set(module_rows) - {module_id}) + if duplicate_module_ids or unexpected_modules: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_manifest_set_mismatch", + code="installed_manifest_set_mismatch", + message=f"Installed distribution {package_name!r} exposes an unexpected or duplicate module entry-point set.", + ) + + record = artifact.get("record_integrity") + record_status = record.get("status") if isinstance(record, dict) else None + if record_status == "verified": + record_verified += 1 + else: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_record_unverified", + code="installed_record_integrity_unverified", + message=f"Installed distribution {package_name!r} does not have complete, matching RECORD hash coverage.", + extra={"record_status": record_status}, + ) + + provenance = artifact.get("source_provenance") + provenance_kind = ( + str(provenance.get("kind") or "") if isinstance(provenance, dict) else "" + ) + if provenance_kind == "vcs-commit": + commit = str(provenance.get("commit") or "").lower() + selected_version = selected_versions.get(repository) + selected_commit = selected_commits.get(repository) + assessed_commit = str(component.get("commit") or "").lower() + commit_matches = commit.startswith(assessed_commit) + selected_matches = ( + selected_version != expected_version + or selected_commit is None + or commit == selected_commit.lower() + ) + if commit_matches and selected_matches: + provenance_anchored += 1 + else: + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_commit_mismatch", + code="installed_source_commit_mismatch", + message=f"Installed immutable VCS provenance for {package_name!r} does not match the assessed or signed selected-unit commit.", + ) + elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}: + provenance_mutable += 1 + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_source_mutable", + code="installed_source_mutable", + message=f"Installed distribution {package_name!r} is editable or directory-backed and is not immutable release evidence.", + ) + else: + provenance_unanchored += 1 + artifact_valid = False + _record_change( + findings=findings, + changes=changes, + changed_repositories=changed_repositories, + affected_module_ids=affected_module_ids, + component=component, + kind="installed_source_unanchored", + code="installed_source_unanchored", + message=f"Installed distribution {package_name!r} has no source provenance anchored to the assessed release.", + ) + + expected_packages = set(expected_by_package) + for package_name in sorted(set(artifacts_by_package) - expected_packages): + artifact_valid = False + matching_module_id = next( + ( + module_id + for module_id, entry in catalog_entries.items() + if normalize_package_name(str(entry.get("package") or "")) + == package_name + ), + None, + ) + component = assessment_by_module.get(str(matching_module_id or "")) + repository = ( + str(component.get("repository")) if isinstance(component, dict) else None + ) + if repository: + changed_repositories.add(repository) + if matching_module_id and component is not None: + affected_module_ids.add(str(matching_module_id)) + target_ids = ( + (f"composition.{matching_module_id}",) + if matching_module_id and component is not None + else ("assessment.installed_composition",) + ) + findings.append( + EvidenceFinding( + "review", + "installed_distribution_extra", + f"Installed GovOPlaN distribution {package_name!r} is outside the enabled assessed composition.", + target_ids, + ) + ) + changes.append( + { + "module_id": str(matching_module_id or package_name), + "repository": repository, + "kind": "installed_distribution_extra", + "package_name": package_name, + } + ) + + for issue in evidence.get("collection_issues", []): + if not isinstance(issue, dict): + continue + package_name = str(issue.get("package_name") or "") + code = str(issue.get("code") or "") + artifact_valid = False + findings.append( + EvidenceFinding( + "review", + "installed_collection_incomplete", + f"Installed evidence collection reported {code!r} for {package_name!r}.", + ("assessment.installed_composition",), + ) + ) + + exact_set = ( + set(artifacts_by_package) == expected_packages and not duplicate_packages + ) + installed_checked = True + record_checked = binding_valid and exact_set and record_expected > 0 + provenance_checked = binding_valid and exact_set and provenance_expected > 0 + proof_scope = { + "installed_artifacts": { + "checked": installed_checked, + "valid": artifact_valid, + "evidence_sha256": digest, + "expected_distribution_count": len(expected_packages), + "observed_distribution_count": len(artifact_rows), + }, + "installed_record_integrity": { + "checked": record_checked, + "valid": record_verified == record_expected if record_checked else None, + "verified_distribution_count": record_verified, + "expected_distribution_count": record_expected, + }, + "installed_source_provenance": { + "checked": provenance_checked, + "valid": provenance_anchored == provenance_expected + if provenance_checked + else None, + "anchored_distribution_count": provenance_anchored, + "mutable_distribution_count": provenance_mutable, + "unanchored_distribution_count": provenance_unanchored, + "expected_distribution_count": provenance_expected, + }, + "runtime_activation": { + "checked": False, + "valid": None, + "required_evidence": ( + "A separately collected runtime registry/configuration snapshot bound to the installed evidence digest.", + ), + }, + } + return InstalledEvidenceReview( + findings=tuple(findings), + changes=tuple(changes), + changed_repositories=frozenset(changed_repositories), + affected_module_ids=frozenset(affected_module_ids), + proof_scope=proof_scope, + evidence_sha256=digest, + ) + + +def review_boundary_evidence( + *, + assessment: dict[str, Any], + installed_review: InstalledEvidenceReview, + evidence: dict[str, Any] | None, + evidence_schema: dict[str, Any] | None, + authority_keyring: dict[str, Any] | None, + authority_keyring_schema: dict[str, Any] | None, + verification_time: datetime | None = None, +) -> BoundaryEvidenceReview: + scopes = _unchecked_boundary_scope() + if evidence is None: + return BoundaryEvidenceReview(findings=(), proof_scope=scopes) + findings: list[EvidenceFinding] = [] + if evidence_schema is None or authority_keyring_schema is None: + findings.append( + EvidenceFinding( + "blocker", + "boundary_evidence_schema_missing", + "Boundary evidence was supplied without both validation schemas.", + ("assessment.external_proof",), + ) + ) + return BoundaryEvidenceReview(tuple(findings), scopes) + evidence_errors = validate_payload(payload=evidence, schema=evidence_schema) + keyring_errors = ( + validate_payload(payload=authority_keyring, schema=authority_keyring_schema) + if isinstance(authority_keyring, dict) + else ("$: an independently provisioned proof-authority keyring is required",) + ) + findings.extend( + EvidenceFinding( + "blocker", + "boundary_evidence_schema", + message, + ("assessment.external_proof",), + ) + for message in evidence_errors + ) + findings.extend( + EvidenceFinding( + "blocker", + "boundary_authority_keyring_schema", + message, + ("assessment.external_proof",), + ) + for message in keyring_errors + ) + if evidence_errors or keyring_errors or not isinstance(authority_keyring, dict): + return BoundaryEvidenceReview(tuple(findings), scopes) + + expected_release = ( + assessment.get("release", {}).get("ref") + if isinstance(assessment.get("release"), dict) + else None + ) + if evidence.get("assessment_id") != assessment.get("assessment_id"): + findings.append( + EvidenceFinding( + "blocker", + "boundary_assessment_mismatch", + "Boundary evidence is bound to a different assessment ID.", + ("assessment.external_proof",), + ) + ) + if evidence.get("assessment_release") != expected_release: + findings.append( + EvidenceFinding( + "blocker", + "boundary_release_mismatch", + "Boundary evidence is bound to a different assessment release.", + ("assessment.release", "assessment.external_proof"), + ) + ) + installed_digest = installed_review.evidence_sha256 + installed_scope = installed_review.proof_scope.get("installed_artifacts", {}) + if ( + installed_digest is None + or evidence.get("installed_evidence_sha256") != installed_digest + or installed_scope.get("valid") is not True + ): + findings.append( + EvidenceFinding( + "blocker", + "boundary_installed_evidence_untrusted", + "Boundary evidence must bind to the supplied, valid installed-composition evidence digest.", + ("assessment.installed_composition", "assessment.external_proof"), + ) + ) + + now = verification_time or datetime.now(tz=UTC) + if now.tzinfo is None: + now = now.replace(tzinfo=UTC) + issued_at = _datetime(evidence.get("issued_at")) + expires_at = _datetime(evidence.get("expires_at")) + time_valid = True + if issued_at is None or expires_at is None or issued_at >= expires_at: + time_valid = False + findings.append( + EvidenceFinding( + "blocker", + "boundary_evidence_interval_invalid", + "Boundary evidence must have issued_at earlier than expires_at.", + ("assessment.external_proof",), + ) + ) + elif now < issued_at or now >= expires_at: + time_valid = False + findings.append( + EvidenceFinding( + "blocker", + "boundary_evidence_not_current", + "Boundary evidence is not valid at the verification time.", + ("assessment.external_proof",), + ) + ) + + claims = [item for item in evidence.get("claims", []) if isinstance(item, dict)] + claims_by_scope: dict[str, dict[str, Any]] = {} + duplicate_scopes: set[str] = set() + semantic_invalid: set[str] = set() + for claim in claims: + scope = str(claim.get("scope") or "") + if scope in claims_by_scope: + duplicate_scopes.add(scope) + else: + claims_by_scope[scope] = claim + allowed_results = {POSITIVE_RESULTS.get(scope), NEGATIVE_RESULTS.get(scope)} + if claim.get("result") not in allowed_results: + semantic_invalid.add(scope) + controls = claim.get("control_ids") + if not isinstance(controls, list) or len(controls) != len(set(controls)): + semantic_invalid.add(scope) + artifacts = claim.get("artifacts") + artifact_ids = [ + str(item.get("artifact_id") or "") + for item in artifacts or [] + if isinstance(item, dict) + ] + if len(artifact_ids) != len(set(artifact_ids)): + semantic_invalid.add(scope) + for scope in sorted(duplicate_scopes): + findings.append( + EvidenceFinding( + "blocker", + "boundary_claim_duplicate", + f"Boundary evidence repeats claim scope {scope!r}.", + ("assessment.external_proof",), + ) + ) + for scope in sorted(semantic_invalid): + findings.append( + EvidenceFinding( + "blocker", + "boundary_claim_semantics", + f"Boundary claim {scope!r} has an invalid result or duplicate control/artifact IDs.", + ("assessment.external_proof",), + ) + ) + + authority_keys, key_findings = _active_authority_keys( + authority_keyring=authority_keyring, + verification_time=now, + ) + findings.extend(key_findings) + signature_payload = dict(evidence) + signature_payload.pop("signatures", None) + signature_bytes = canonical_bytes(signature_payload) + authorized_scopes: set[str] = set() + seen_signature_key_ids: set[str] = set() + duplicate_signature_key_ids: set[str] = set() + for signature in evidence.get("signatures", []): + if not isinstance(signature, dict): + continue + key_id = str(signature.get("key_id") or "") + if key_id in seen_signature_key_ids: + duplicate_signature_key_ids.add(key_id) + continue + seen_signature_key_ids.add(key_id) + key = authority_keys.get(key_id) + if key is None or signature.get("algorithm") != "ed25519": + continue + try: + signature_value = base64.b64decode( + str(signature.get("value") or ""), validate=True + ) + Ed25519PublicKey.from_public_bytes(key[0]).verify( + signature_value, signature_bytes + ) + except (ValueError, InvalidSignature): + continue + authorized_scopes.update(key[1]) + if duplicate_signature_key_ids: + findings.append( + EvidenceFinding( + "blocker", + "boundary_signature_duplicate", + "Boundary evidence repeats one or more signature key IDs.", + ("assessment.external_proof",), + ) + ) + + unauthorized_scopes = { + scope + for scope in claims_by_scope + if scope in PROOF_REQUIREMENTS and scope not in authorized_scopes + } + for scope in sorted(unauthorized_scopes): + findings.append( + EvidenceFinding( + "blocker", + "boundary_authority_untrusted", + f"No valid independently trusted signature is authorized for {scope!r}.", + ("assessment.external_proof",), + ) + ) + global_valid = not any(item.severity == "blocker" for item in findings) + evidence_digest = canonical_sha256(evidence) + for scope, claim in sorted(claims_by_scope.items()): + if scope not in PROOF_REQUIREMENTS: + continue + if scope in unauthorized_scopes: + continue + if scope in duplicate_scopes or scope in semantic_invalid or not time_valid: + continue + if not global_valid: + continue + result = str(claim.get("result")) + valid = result == POSITIVE_RESULTS[scope] + scopes[scope] = { + "checked": True, + "valid": valid, + "evidence_sha256": evidence_digest, + "result": result, + "control_count": len(claim.get("control_ids") or []), + "artifact_count": len(claim.get("artifacts") or []), + "required_evidence": PROOF_REQUIREMENTS[scope], + } + if not valid: + findings.append( + EvidenceFinding( + "review", + "boundary_claim_negative", + f"Authorized boundary claim {scope!r} has result {result!r}.", + ("assessment.external_proof",), + ) + ) + return BoundaryEvidenceReview(tuple(findings), scopes) + + +def normalize_package_name(value: str) -> str: + return re.sub(r"[-_.]+", "-", value.strip().lower()) + + +def canonical_bytes(payload: object) -> bytes: + return json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + + +def canonical_sha256(payload: object) -> str: + return hashlib.sha256(canonical_bytes(payload)).hexdigest() + + +def _distribution_version(distribution: metadata.Distribution) -> str: + try: + return str(distribution.version) + except Exception: + return "" + + +def _distribution_modules( + *, package_name: str, distribution: metadata.Distribution +) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + modules: list[dict[str, str]] = [] + issues: list[dict[str, str]] = [] + try: + all_entry_points = sorted( + ( + entry_point + for entry_point in distribution.entry_points + if entry_point.group == "govoplan.modules" + ), + key=lambda item: (item.name, item.value), + ) + except Exception: + return [], [ + { + "code": "distribution-metadata-invalid", + "package_name": package_name, + } + ] + if len(all_entry_points) > 16: + issues.append( + { + "code": "module-entry-point-limit-exceeded", + "package_name": package_name, + } + ) + for entry_point in all_entry_points[:16]: + try: + manifest = entry_point.load()() + module_id = str(manifest.id) + manifest_version = str(manifest.version) + except (Exception, SystemExit): + issues.append( + { + "code": "module-entry-point-load-failed", + "package_name": package_name, + } + ) + continue + if ( + OPAQUE_ID_PATTERN.fullmatch(module_id) is None + or VERSION_PATTERN.fullmatch(manifest_version) is None + or len(manifest_version) > 128 + ): + issues.append( + { + "code": "module-entry-point-invalid", + "package_name": package_name, + } + ) + continue + modules.append({"module_id": module_id, "manifest_version": manifest_version}) + return sorted(modules, key=lambda item: item["module_id"]), issues + + +def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: + try: + raw = distribution.read_text("direct_url.json") + except Exception: + return {"kind": "malformed-direct-url"} + if raw is None: + return {"kind": "index-or-unknown"} + if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES: + return {"kind": "malformed-direct-url"} + try: + direct_url = json.loads(raw) + except (TypeError, json.JSONDecodeError): + return {"kind": "malformed-direct-url"} + if not isinstance(direct_url, dict): + return {"kind": "malformed-direct-url"} + vcs_info = direct_url.get("vcs_info") + directory_info = direct_url.get("dir_info") + editable = ( + isinstance(directory_info, dict) and directory_info.get("editable") is True + ) + if isinstance(vcs_info, dict): + commit = str(vcs_info.get("commit_id") or "").lower() + if GIT_OBJECT_PATTERN.fullmatch(commit): + return { + "kind": "editable-vcs" if editable else "vcs-commit", + "commit": commit, + } + return {"kind": "malformed-direct-url"} + archive_info = direct_url.get("archive_info") + if isinstance(archive_info, dict): + hashes = archive_info.get("hashes") + digest = hashes.get("sha256") if isinstance(hashes, dict) else None + if digest is None: + legacy = str(archive_info.get("hash") or "") + if legacy.startswith("sha256="): + digest = legacy.removeprefix("sha256=") + digest_text = str(digest or "").lower() + if SHA256_PATTERN.fullmatch(digest_text): + return {"kind": "archive", "sha256": digest_text} + return {"kind": "malformed-direct-url"} + if isinstance(directory_info, dict): + return {"kind": "editable-local" if editable else "local-directory"} + return {"kind": "malformed-direct-url"} + + +def _record_integrity( + distribution: metadata.Distribution, + *, + budget: CollectionHashBudget, +) -> dict[str, Any]: + counts = { + "hashed_file_count": 0, + "permitted_unhashed_file_count": 0, + "unverifiable_file_count": 0, + "missing_file_count": 0, + "mismatched_file_count": 0, + } + try: + files = tuple(distribution.files or ()) + except Exception: + files = () + if not files: + return {"status": "unavailable", **counts} + if len(files) > MAX_DISTRIBUTION_FILES or len(files) > budget.remaining_files: + budget.remaining_files = 0 + return {"status": "limit-exceeded", **counts} + budget.remaining_files -= len(files) + try: + distribution_root = Path(distribution.locate_file("")).resolve() + installation_root = Path(sys.prefix).resolve() + except OSError: + return {"status": "unavailable", **counts} + total_hashed_bytes = 0 + limit_exceeded = False + for package_path in files: + hash_info = getattr(package_path, "hash", None) + if hash_info is None: + if _permitted_unhashed_record_path(str(package_path)): + counts["permitted_unhashed_file_count"] += 1 + else: + counts["unverifiable_file_count"] += 1 + continue + if getattr(hash_info, "mode", None) != "sha256": + counts["unverifiable_file_count"] += 1 + continue + try: + resolved_path = Path(distribution.locate_file(package_path)).resolve() + except OSError: + counts["missing_file_count"] += 1 + continue + if not ( + _is_relative_to(resolved_path, distribution_root) + or _is_relative_to(resolved_path, installation_root) + ): + counts["unverifiable_file_count"] += 1 + continue + try: + stat = resolved_path.stat() + except OSError: + counts["missing_file_count"] += 1 + continue + if not resolved_path.is_file(): + counts["missing_file_count"] += 1 + continue + if stat.st_size > MAX_HASHED_FILE_BYTES: + limit_exceeded = True + continue + total_hashed_bytes += stat.st_size + if ( + total_hashed_bytes > MAX_HASHED_DISTRIBUTION_BYTES + or stat.st_size > budget.remaining_bytes + ): + limit_exceeded = True + budget.remaining_bytes = 0 + continue + budget.remaining_bytes -= stat.st_size + expected = _record_hash_bytes(str(getattr(hash_info, "value", ""))) + if expected is None: + counts["unverifiable_file_count"] += 1 + continue + digest = hashlib.sha256() + try: + with resolved_path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + except OSError: + counts["missing_file_count"] += 1 + continue + counts["hashed_file_count"] += 1 + if not hmac.compare_digest(digest.digest(), expected): + counts["mismatched_file_count"] += 1 + if limit_exceeded: + status = "limit-exceeded" + elif counts["missing_file_count"] or counts["mismatched_file_count"]: + status = "mismatch" + elif counts["hashed_file_count"] == 0: + status = "unavailable" + elif counts["unverifiable_file_count"]: + status = "partial" + else: + status = "verified" + return {"status": status, **counts} + + +def _record_hash_bytes(value: str) -> bytes | None: + try: + decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + except (ValueError, TypeError): + return None + return decoded if len(decoded) == hashlib.sha256().digest_size else None + + +def _permitted_unhashed_record_path(value: str) -> bool: + normalized = value.replace("\\", "/") + return normalized.endswith( + (".dist-info/RECORD", ".dist-info/RECORD.jws", ".dist-info/RECORD.p7s") + ) + + +def _active_authority_keys( + *, authority_keyring: dict[str, Any], verification_time: datetime +) -> tuple[dict[str, tuple[bytes, frozenset[str]]], tuple[EvidenceFinding, ...]]: + result: dict[str, tuple[bytes, frozenset[str]]] = {} + findings: list[EvidenceFinding] = [] + seen: set[str] = set() + for item in authority_keyring.get("keys", []): + if not isinstance(item, dict): + continue + key_id = str(item.get("key_id") or "") + if key_id in seen: + findings.append( + EvidenceFinding( + "blocker", + "boundary_authority_key_duplicate", + "Proof-authority keyring repeats a key ID.", + ("assessment.external_proof",), + ) + ) + continue + seen.add(key_id) + if item.get("status") not in {"active", "next"}: + continue + not_before = _datetime(item.get("not_before")) + not_after = _datetime(item.get("not_after")) + if (not_before is not None and verification_time < not_before) or ( + not_after is not None and verification_time > not_after + ): + continue + try: + public_key = base64.b64decode( + str(item.get("public_key") or ""), validate=True + ) + Ed25519PublicKey.from_public_bytes(public_key) + except ValueError: + continue + result[key_id] = ( + public_key, + frozenset(str(value) for value in item.get("allowed_scopes", [])), + ) + return result, tuple(findings) + + +def _record_change( + *, + findings: list[EvidenceFinding], + changes: list[dict[str, Any]], + changed_repositories: set[str], + affected_module_ids: set[str], + component: Mapping[str, Any], + kind: str, + code: str, + message: str, + extra: Mapping[str, Any] | None = None, +) -> None: + module_id = str(component.get("module_id")) + repository = str(component.get("repository")) + changed_repositories.add(repository) + affected_module_ids.add(module_id) + findings.append( + EvidenceFinding( + "review", + code, + message, + (f"composition.{module_id}",), + ) + ) + change: dict[str, Any] = { + "module_id": module_id, + "repository": repository, + "kind": kind, + } + if extra: + change.update(extra) + changes.append(change) + + +def _unchecked_installed_scope() -> dict[str, Any]: + return { + "installed_artifacts": { + "checked": False, + "valid": None, + "required_evidence": ( + "A bounded installed-composition evidence document collected from the Python environment under review.", + ), + }, + "installed_record_integrity": {"checked": False, "valid": None}, + "installed_source_provenance": {"checked": False, "valid": None}, + "runtime_activation": { + "checked": False, + "valid": None, + "required_evidence": ( + "A separately collected runtime registry/configuration snapshot bound to installed evidence.", + ), + }, + } + + +def _invalid_installed_scope(*, digest: str) -> dict[str, Any]: + scope = _unchecked_installed_scope() + scope["installed_artifacts"] = { + "checked": True, + "valid": False, + "evidence_sha256": digest, + } + return scope + + +def _unchecked_boundary_scope() -> dict[str, Any]: + return { + scope: { + "checked": False, + "valid": None, + "required_evidence": requirements, + } + for scope, requirements in PROOF_REQUIREMENTS.items() + } + + +def _datetime(value: object) -> datetime | None: + if not isinstance(value, str) or not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + + +def _json_path(parts: Iterable[object]) -> str: + path = "$" + for part in parts: + path += f"[{part}]" if isinstance(part, int) else f".{part}" + return path + + +def _unique_dicts(values: Sequence[dict[str, str]]) -> list[dict[str, str]]: + result: list[dict[str, str]] = [] + seen: set[tuple[tuple[str, str], ...]] = set() + for value in values: + key = tuple(sorted(value.items())) + if key not in seen: + seen.add(key) + result.append(value) + return result + + +def _is_relative_to(path: Path, root: Path) -> bool: + try: + path.relative_to(root) + except ValueError: + return False + return True