diff --git a/docs/installed-composition-evidence.schema.json b/docs/installed-composition-evidence.schema.json index 9e607ca..88ce481 100644 --- a/docs/installed-composition-evidence.schema.json +++ b/docs/installed-composition-evidence.schema.json @@ -20,7 +20,7 @@ "format": "uri-reference" }, "schema_version": { - "const": "0.3.0" + "const": "0.4.0" }, "evidence_kind": { "const": "govoplan.installed-composition" @@ -182,7 +182,9 @@ "generated_unhashed_file_count", "unverifiable_file_count", "missing_file_count", - "mismatched_file_count" + "mismatched_file_count", + "artifact_payload_identity", + "installed_payload_identity" ], "properties": { "status": { @@ -211,6 +213,18 @@ }, "mismatched_file_count": { "$ref": "#/$defs/count" + }, + "artifact_payload_identity": { + "oneOf": [ + { "$ref": "#/$defs/artifact_payload_identity" }, + { "type": "null" } + ] + }, + "installed_payload_identity": { + "oneOf": [ + { "$ref": "#/$defs/installed_payload_identity" }, + { "type": "null" } + ] } }, "allOf": [ @@ -222,7 +236,9 @@ "permitted_unhashed_file_count": { "minimum": 1 }, "unverifiable_file_count": { "const": 0 }, "missing_file_count": { "const": 0 }, - "mismatched_file_count": { "const": 0 } + "mismatched_file_count": { "const": 0 }, + "artifact_payload_identity": { "$ref": "#/$defs/artifact_payload_identity" }, + "installed_payload_identity": { "$ref": "#/$defs/installed_payload_identity" } } } }, @@ -278,6 +294,34 @@ } } }, + "artifact_payload_identity": { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "sha256", "file_count"], + "properties": { + "algorithm": { "const": "govoplan-wheel-declared-payload-v1" }, + "sha256": { "$ref": "#/$defs/sha256" }, + "file_count": { + "type": "integer", + "minimum": 1, + "maximum": 10000 + } + } + }, + "installed_payload_identity": { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "sha256", "file_count"], + "properties": { + "algorithm": { "const": "govoplan-installed-record-payload-v1" }, + "sha256": { "$ref": "#/$defs/sha256" }, + "file_count": { + "type": "integer", + "minimum": 1, + "maximum": 10000 + } + } + }, "count": { "type": "integer", "minimum": 0, diff --git a/docs/installer-receipt-authority-keyring.schema.json b/docs/installer-receipt-authority-keyring.schema.json new file mode 100644 index 0000000..035ed73 --- /dev/null +++ b/docs/installer-receipt-authority-keyring.schema.json @@ -0,0 +1,53 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.add-ideas.de/add-ideas/govoplan/src/branch/main/docs/installer-receipt-authority-keyring.schema.json", + "title": "GovOPlaN installer receipt 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.installer-receipt-authorities" }, + "keys": { + "type": "array", + "minItems": 1, + "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": 1, + "uniqueItems": true, + "items": { "const": "installed_release_origin" } + }, + "not_before": { "type": "string", "format": "date-time" }, + "not_after": { "type": "string", "format": "date-time" } + } + } + } +} diff --git a/docs/installer-receipt.schema.json b/docs/installer-receipt.schema.json new file mode 100644 index 0000000..c72a0c8 --- /dev/null +++ b/docs/installer-receipt.schema.json @@ -0,0 +1,122 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.add-ideas.de/add-ideas/govoplan/src/branch/main/docs/installer-receipt.schema.json", + "title": "GovOPlaN signed installer receipt", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "evidence_kind", + "receipt_id", + "assessment_id", + "assessment_release", + "installed_evidence_sha256", + "catalog", + "issued_at", + "artifacts", + "signatures" + ], + "properties": { + "$schema": { "type": "string", "format": "uri-reference" }, + "schema_version": { "const": "0.1.0" }, + "evidence_kind": { "const": "govoplan.installer-receipt" }, + "receipt_id": { "$ref": "#/$defs/opaque_id" }, + "assessment_id": { "$ref": "#/$defs/opaque_id" }, + "assessment_release": { "$ref": "#/$defs/opaque_id" }, + "installed_evidence_sha256": { "$ref": "#/$defs/sha256" }, + "catalog": { + "type": "object", + "additionalProperties": false, + "required": ["channel", "sequence", "sha256"], + "properties": { + "channel": { + "type": "string", + "pattern": "^[a-z][a-z0-9_-]{0,63}$" + }, + "sequence": { "type": "integer", "minimum": 1 }, + "sha256": { "$ref": "#/$defs/sha256" } + } + }, + "issued_at": { "type": "string", "format": "date-time" }, + "artifacts": { + "type": "array", + "minItems": 1, + "maxItems": 256, + "items": { "$ref": "#/$defs/artifact" } + }, + "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._:-]*$" + }, + "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", + "catalog_archive_sha256", + "installed_payload" + ], + "properties": { + "package_name": { "$ref": "#/$defs/package_name" }, + "package_version": { "$ref": "#/$defs/version" }, + "catalog_archive_sha256": { "$ref": "#/$defs/sha256" }, + "installed_payload": { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "sha256", "file_count"], + "properties": { + "algorithm": { "const": "govoplan-installed-record-payload-v1" }, + "sha256": { "$ref": "#/$defs/sha256" }, + "file_count": { + "type": "integer", + "minimum": 1, + "maximum": 10000 + } + } + } + } + }, + "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/tests/test_capability_fit_evidence.py b/tests/test_capability_fit_evidence.py index a7f3591..948e244 100644 --- a/tests/test_capability_fit_evidence.py +++ b/tests/test_capability_fit_evidence.py @@ -37,6 +37,10 @@ from govoplan_assessment.evidence import ( # noqa: E402 review_installed_composition, validate_payload, ) +from govoplan_assessment.installer_receipt import ( # noqa: E402 + issue_installer_receipt, +) +from govoplan_release.artifact_identity import inspect_python_wheel # noqa: E402 from tests.test_capability_fit_review import signed_catalog # noqa: E402 @@ -89,6 +93,202 @@ class CapabilityFitEvidenceTests(unittest.TestCase): report["proof_scope"]["installed_artifacts"]["evidence_sha256"], ) + def test_signed_catalog_artifact_identities_bind_installed_release_origin( + self, + ) -> None: + evidence = matching_installed_evidence(self.assessment) + catalog, keyring = signed_catalog( + self.assessment, + release_artifacts=catalog_artifacts_for(evidence), + ) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + ) + + origin = report["proof_scope"]["installed_release_origin"] + self.assertEqual("current", report["status"]) + self.assertTrue(origin["checked"]) + self.assertTrue(origin["valid"]) + self.assertTrue(origin["release_origin_bound"]) + self.assertEqual(len(evidence["artifacts"]), origin["anchored_artifact_digest_count"]) + self.assertEqual(0, origin["signed_installer_receipt_count"]) + + def test_role_scoped_installer_receipt_admits_transformed_artifacts_and_boundary( + self, + ) -> None: + evidence = matching_installed_evidence(self.assessment) + artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True) + catalog, keyring = signed_catalog( + self.assessment, + release_artifacts=artifact_rows, + ) + receipt, installer_keys = signed_installer_receipt( + assessment=self.assessment, + installed=evidence, + catalog=catalog, + catalog_artifacts=artifact_rows, + ) + proof, proof_keys = signed_boundary_evidence( + assessment=self.assessment, + installed=evidence, + claims=[boundary_claim("target_environment", "passed")], + allowed_scopes=["target_environment"], + ) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + installer_receipt=receipt, + installer_authority_keyring=installer_keys, + boundary_evidence=proof, + authority_keyring=proof_keys, + installed_evidence_mode="imported_unsigned", + ) + + origin = report["proof_scope"]["installed_release_origin"] + self.assertTrue(origin["valid"]) + self.assertEqual(0, origin["anchored_artifact_digest_count"]) + self.assertEqual(len(evidence["artifacts"]), origin["signed_installer_receipt_count"]) + self.assertTrue(report["proof_scope"]["target_environment"]["checked"]) + self.assertTrue(report["proof_scope"]["target_environment"]["valid"]) + self.assertTrue( + report["proof_scope"]["installed_evidence_observation"][ + "receipt_authenticated" + ] + ) + self.assertNotIn( + "installed_evidence_unsigned_import", + {item["code"] for item in report["findings"]}, + ) + + def test_installer_receipt_cannot_reuse_release_key_or_cover_forged_payload( + self, + ) -> None: + evidence = matching_installed_evidence(self.assessment) + artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True) + catalog, keyring = signed_catalog( + self.assessment, + release_artifacts=artifact_rows, + ) + receipt, installer_keys = signed_installer_receipt( + assessment=self.assessment, + installed=evidence, + catalog=catalog, + catalog_artifacts=artifact_rows, + ) + forged = deepcopy(evidence) + forged["artifacts"][0]["record_integrity"]["installed_payload_identity"][ + "sha256" + ] = "f" * 64 + reused = deepcopy(installer_keys) + reused["keys"][0]["public_key"] = keyring["keys"][0]["public_key"] + + forged_report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=forged, + installer_receipt=receipt, + installer_authority_keyring=installer_keys, + ) + reused_report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + installer_receipt=receipt, + installer_authority_keyring=reused, + ) + + self.assertIn( + "installer_receipt_evidence_mismatch", + {item["code"] for item in forged_report["findings"]}, + ) + self.assertIn( + "installer_authority_untrusted", + {item["code"] for item in reused_report["findings"]}, + ) + self.assertFalse( + forged_report["proof_scope"]["installed_release_origin"]["valid"] + ) + self.assertFalse( + reused_report["proof_scope"]["installed_release_origin"]["valid"] + ) + + def test_receipt_issuer_refuses_file_like_or_unverified_observations(self) -> None: + evidence = matching_installed_evidence(self.assessment) + artifacts = catalog_artifacts_for(evidence, requires_receipt=True) + catalog, _ = signed_catalog( + self.assessment, + release_artifacts=artifacts, + ) + private_key = Ed25519PrivateKey.generate() + + with self.assertRaisesRegex(ValueError, "same-process"): + issue_installer_receipt( + assessment=self.assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id="install:test", + key_id="installer:test", + private_key=private_key, + ) + evidence["artifacts"][0]["record_integrity"]["status"] = "partial" + evidence["artifacts"][0]["record_integrity"]["unverifiable_file_count"] = 1 + with self.assertRaisesRegex(ValueError, "not a verified match"): + issue_installer_receipt( + assessment=self.assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id="install:test", + key_id="installer:test", + private_key=private_key, + same_process_observation=True, + consumed_artifacts={ + str(item["package_name"]): Path("unused.whl") + for item in evidence["artifacts"] + }, + ) + + def test_installer_receipt_and_observation_must_be_current_together(self) -> None: + evidence = matching_installed_evidence(self.assessment) + artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True) + catalog, keyring = signed_catalog( + self.assessment, + release_artifacts=artifact_rows, + ) + receipt, installer_keys = signed_installer_receipt( + assessment=self.assessment, + installed=evidence, + catalog=catalog, + catalog_artifacts=artifact_rows, + ) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + installer_receipt=receipt, + installer_authority_keyring=installer_keys, + installed_evidence_mode="imported_unsigned", + verification_time=datetime(2026, 7, 23, 12, 5, 1, tzinfo=UTC), + ) + + self.assertIn( + "installer_receipt_time_invalid", + {item["code"] for item in report["findings"]}, + ) + self.assertFalse( + report["proof_scope"]["installed_release_origin"]["valid"] + ) + self.assertFalse( + report["proof_scope"]["installed_evidence_observation"][ + "receipt_authenticated" + ] + ) + def test_catalog_trust_blocker_invalidates_dependent_evidence_scopes( self, ) -> None: @@ -523,6 +723,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): text=True, ) wheel = next(wheels.glob("govoplan_core-*.whl")) + built_identity = inspect_python_wheel(wheel) subprocess.run( ( str(build_python), @@ -568,6 +769,15 @@ class CapabilityFitEvidenceTests(unittest.TestCase): self.assertGreater(record["permitted_unhashed_file_count"], 0) self.assertGreater(record["generated_unhashed_file_count"], 0) self.assertEqual(0, record["unverifiable_file_count"]) + self.assertEqual( + built_identity.payload_sha256, + record["artifact_payload_identity"]["sha256"], + ) + self.assertEqual( + built_identity.payload_file_count, + record["artifact_payload_identity"]["file_count"], + ) + self.assertTrue(built_identity.requires_installer_receipt) self.assertEqual( (), validate_payload(payload=evidence, schema=self.installed_schema) ) @@ -1162,6 +1372,8 @@ class CapabilityFitEvidenceTests(unittest.TestCase): catalog: dict[str, object], keyring: dict[str, object], installed_evidence: dict[str, object] | None = None, + installer_receipt: dict[str, object] | None = None, + installer_authority_keyring: dict[str, object] | None = None, boundary_evidence: dict[str, object] | None = None, authority_keyring: dict[str, object] | None = None, installed_evidence_mode: str = "direct_local", @@ -1179,6 +1391,16 @@ class CapabilityFitEvidenceTests(unittest.TestCase): installed_evidence_schema=(installed_schema or self.installed_schema) if installed_evidence is not None else None, + installer_receipt=installer_receipt, + installer_receipt_schema=load_json("docs/installer-receipt.schema.json") + if installer_receipt is not None + else None, + installer_authority_keyring=installer_authority_keyring, + installer_authority_keyring_schema=load_json( + "docs/installer-receipt-authority-keyring.schema.json" + ) + if installer_receipt is not None + else None, boundary_evidence=boundary_evidence, boundary_evidence_schema=self.boundary_schema if boundary_evidence is not None @@ -1196,7 +1418,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]: return { "$schema": "./installed-composition-evidence.schema.json", - "schema_version": "0.3.0", + "schema_version": "0.4.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], @@ -1238,6 +1460,20 @@ def artifact( "unverifiable_file_count": 0, "missing_file_count": 0, "mismatched_file_count": 0, + "artifact_payload_identity": { + "algorithm": "govoplan-wheel-declared-payload-v1", + "sha256": hashlib.sha256( + f"artifact:{package_name}:{version}".encode() + ).hexdigest(), + "file_count": 1, + }, + "installed_payload_identity": { + "algorithm": "govoplan-installed-record-payload-v1", + "sha256": hashlib.sha256( + f"installed:{package_name}:{version}".encode() + ).hexdigest(), + "file_count": 1, + }, }, } @@ -1323,6 +1559,107 @@ def signed_boundary_evidence( return proof, keyring +def catalog_artifacts_for( + installed: dict[str, object], *, requires_receipt: bool = False +) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for item in installed["artifacts"]: + record = item["record_integrity"] + rows.append( + { + "artifact_kind": "python-wheel", + "package_name": item["package_name"], + "package_version": item["package_version"], + "archive_sha256": hashlib.sha256( + f"wheel:{item['package_name']}:{item['package_version']}".encode() + ).hexdigest(), + "archive_size": 1024, + "installed_payload": deepcopy(record["artifact_payload_identity"]), + "requires_installer_receipt": requires_receipt, + } + ) + return rows + + +def signed_installer_receipt( + *, + assessment: dict[str, object], + installed: dict[str, object], + catalog: dict[str, object], + catalog_artifacts: list[dict[str, object]], +) -> 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") + catalog_by_package = { + str(item["package_name"]): item for item in catalog_artifacts + } + receipt: dict[str, object] = { + "$schema": "./installer-receipt.schema.json", + "schema_version": "0.1.0", + "evidence_kind": "govoplan.installer-receipt", + "receipt_id": "install:test", + "assessment_id": assessment["assessment_id"], + "assessment_release": assessment["release"]["ref"], + "installed_evidence_sha256": canonical_sha256(installed), + "catalog": { + "channel": catalog["channel"], + "sequence": catalog["sequence"], + "sha256": hashlib.sha256( + json.dumps( + catalog, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + ).hexdigest(), + }, + "issued_at": "2026-07-23T11:59:00Z", + "artifacts": [ + { + "package_name": item["package_name"], + "package_version": item["package_version"], + "catalog_archive_sha256": catalog_by_package[ + str(item["package_name"]) + ]["archive_sha256"], + "installed_payload": deepcopy( + item["record_integrity"]["installed_payload_identity"] + ), + } + for item in installed["artifacts"] + ], + } + receipt["signatures"] = [ + { + "algorithm": "ed25519", + "key_id": "installer:test", + "value": base64.b64encode( + private_key.sign(canonical_bytes(receipt)) + ).decode("ascii"), + } + ] + keyring = { + "$schema": "./installer-receipt-authority-keyring.schema.json", + "schema_version": "0.1.0", + "purpose": "govoplan.installer-receipt-authorities", + "keys": [ + { + "key_id": "installer:test", + "status": "active", + "public_key": public_key, + "allowed_scopes": ["installed_release_origin"], + "not_before": "2026-07-01T00:00:00Z", + "not_after": "2026-08-01T00:00:00Z", + } + ], + } + return receipt, keyring + + def create_distribution( root: Path, ) -> tuple[metadata.PathDistribution, Path]: diff --git a/tests/test_capability_fit_review.py b/tests/test_capability_fit_review.py index 7555a8d..cc69b61 100644 --- a/tests/test_capability_fit_review.py +++ b/tests/test_capability_fit_review.py @@ -606,6 +606,7 @@ def signed_catalog( sequence: int = 202607220843, selected_units: list[dict[str, object]] | None = None, include_selected_units: bool = True, + release_artifacts: list[dict[str, object]] | None = None, ) -> tuple[dict[str, object], dict[str, object]]: versions = versions or {} private_key = Ed25519PrivateKey.generate() @@ -661,6 +662,8 @@ def signed_catalog( release: dict[str, object] = {"keyring_sha256": canonical_hash(keyring)} if include_selected_units: release["selected_units"] = selected_units + if release_artifacts is not None: + release["artifacts"] = release_artifacts catalog: dict[str, object] = { "catalog_version": "1", "channel": "stable", diff --git a/tests/test_release_artifact_identity.py b/tests/test_release_artifact_identity.py new file mode 100644 index 0000000..95f6c8c --- /dev/null +++ b/tests/test_release_artifact_identity.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +import csv +from datetime import UTC, datetime +import io +from pathlib import Path +import sys +import tempfile +import unittest +from unittest import mock +import zipfile + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +META_ROOT = Path(__file__).resolve().parents[1] +RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" +if str(RELEASE_TOOLS_ROOT) not in sys.path: + sys.path.insert(0, str(RELEASE_TOOLS_ROOT)) + +from govoplan_release.artifact_identity import ( # noqa: E402 + ArtifactIdentityError, + inspect_python_wheel, + selected_artifact_identity_issues, +) +from govoplan_release.selective_catalog import ( # noqa: E402 + apply_python_artifact_identities, +) +ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" +if str(ASSESSMENT_TOOLS_ROOT) not in sys.path: + sys.path.insert(0, str(ASSESSMENT_TOOLS_ROOT)) +from govoplan_assessment.installer_receipt import issue_installer_receipt # noqa: E402 + + +class ReleaseArtifactIdentityTests(unittest.TestCase): + def test_built_wheel_bytes_become_catalog_identity(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + wheel = self._wheel(Path(temp_dir), console_script=True) + payload = self._catalog_payload() + + changes = apply_python_artifact_identities( + payload, + repo_versions={"govoplan-demo": "1.2.3"}, + python_artifacts={"govoplan-demo": wheel}, + ) + + identity = payload["release"]["artifacts"][0] + self.assertEqual("govoplan-demo", identity["package_name"]) + self.assertEqual("1.2.3", identity["package_version"]) + self.assertRegex(identity["archive_sha256"], r"^[0-9a-f]{64}$") + self.assertTrue(identity["requires_installer_receipt"]) + self.assertEqual("release.artifact", changes[0].field) + self.assertEqual((), selected_artifact_identity_issues(payload)) + + def test_version_update_without_wheel_removes_stale_identity_and_blocks_publish(self) -> None: + payload = self._catalog_payload() + payload["release"]["artifacts"] = [ + { + "artifact_kind": "python-wheel", + "package_name": "govoplan-demo", + "package_version": "1.2.2", + "archive_sha256": "a" * 64, + "archive_size": 10, + "installed_payload": { + "algorithm": "govoplan-wheel-declared-payload-v1", + "sha256": "b" * 64, + "file_count": 1, + }, + "requires_installer_receipt": False, + } + ] + + apply_python_artifact_identities( + payload, + repo_versions={"govoplan-demo": "1.2.3"}, + python_artifacts={}, + ) + + self.assertNotIn("artifacts", payload["release"]) + self.assertIn( + "no built artifact identity", + " ".join(selected_artifact_identity_issues(payload)), + ) + + def test_tag_only_selection_needs_no_python_artifact_mapping(self) -> None: + payload = self._catalog_payload() + payload["release"]["selected_units"].append( + {"repo": "govoplan", "version": "1.2.3"} + ) + + apply_python_artifact_identities( + payload, + repo_versions={"govoplan": "1.2.3"}, + python_artifacts={}, + ) + + self.assertIn("selected_units", payload["release"]) + + def test_unsafe_or_mismatched_wheel_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + unsafe = root / "govoplan_demo-1.2.3-py3-none-any.whl" + with zipfile.ZipFile(unsafe, "w") as archive: + archive.writestr("../escape", b"bad") + archive.writestr( + "govoplan_demo-1.2.3.dist-info/METADATA", + "Metadata-Version: 2.1\nName: govoplan-demo\nVersion: 1.2.3\n", + ) + wrong = self._wheel(root, version="9.9.9") + + with self.assertRaisesRegex(ArtifactIdentityError, "unsafe"): + inspect_python_wheel(unsafe) + with self.assertRaisesRegex(ValueError, "expected govoplan-demo 1.2.3"): + apply_python_artifact_identities( + self._catalog_payload(), + repo_versions={"govoplan-demo": "1.2.3"}, + python_artifacts={"govoplan-demo": wrong}, + ) + + def test_receipt_issuer_hashes_exact_consumed_wheel(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + original = self._wheel(root / "original", value=b"VALUE = 1\n") + different = self._wheel(root / "different", value=b"VALUE = 2\n") + identity = inspect_python_wheel(original) + catalog = self._catalog_payload() + catalog["channel"] = "stable" + catalog["sequence"] = 1 + catalog["release"]["artifacts"] = [identity.catalog_payload()] + assessment = { + "assessment_id": "assessment:test", + "release": {"ref": "stable-catalog-1"}, + "composition": [{"module_id": "demo", "enabled": True}], + } + installed_payload = { + "algorithm": "govoplan-installed-record-payload-v1", + "sha256": "c" * 64, + "file_count": identity.payload_file_count, + } + evidence = { + "assessment_id": "assessment:test", + "assessment_release": "stable-catalog-1", + "collected_at": "2026-07-22T12:00:00Z", + "artifacts": [ + { + "package_name": "govoplan-demo", + "package_version": "1.2.3", + "record_integrity": { + "status": "verified", + "hashed_file_count": identity.payload_file_count, + "permitted_unhashed_file_count": 1, + "generated_unhashed_file_count": 0, + "unverifiable_file_count": 0, + "missing_file_count": 0, + "mismatched_file_count": 0, + "artifact_payload_identity": catalog["release"]["artifacts"][0]["installed_payload"], + "installed_payload_identity": installed_payload, + }, + } + ], + } + key = Ed25519PrivateKey.generate() + + receipt = issue_installer_receipt( + assessment=assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id="install:test", + key_id="installer:test", + private_key=key, + consumed_artifacts={"govoplan-demo": original}, + issued_at=datetime(2026, 7, 22, 12, 1, tzinfo=UTC), + same_process_observation=True, + ) + with self.assertRaisesRegex(ValueError, "differs from the signed"): + issue_installer_receipt( + assessment=assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id="install:test", + key_id="installer:test", + private_key=key, + consumed_artifacts={"govoplan-demo": different}, + issued_at=datetime(2026, 7, 22, 12, 1, tzinfo=UTC), + same_process_observation=True, + ) + with self.assertRaisesRegex(ValueError, "within five minutes"): + issue_installer_receipt( + assessment=assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id="install:test", + key_id="installer:test", + private_key=key, + consumed_artifacts={"govoplan-demo": original}, + issued_at=datetime(2026, 7, 22, 12, 6, tzinfo=UTC), + same_process_observation=True, + ) + + self.assertEqual(identity.archive_sha256, receipt["artifacts"][0]["catalog_archive_sha256"]) + + def test_path_replacement_during_one_descriptor_inspection_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + target = self._wheel(root / "target", value=b"VALUE = 1\n") + replacement = self._wheel(root / "replacement", value=b"VALUE = 2\n") + real_zip = zipfile.ZipFile + + def swap_path(file_or_path, *args, **kwargs): + target.unlink() + replacement.rename(target) + return real_zip(file_or_path, *args, **kwargs) + + with ( + mock.patch( + "govoplan_release.artifact_identity.zipfile.ZipFile", + side_effect=swap_path, + ), + self.assertRaisesRegex(ArtifactIdentityError, "changed"), + ): + inspect_python_wheel(target) + + @staticmethod + def _catalog_payload() -> dict[str, object]: + return { + "core_release": {}, + "modules": [ + { + "module_id": "demo", + "version": "1.2.3", + "python_package": "govoplan-demo", + "python_ref": "govoplan-demo @ git+ssh://git@example.test/acme/govoplan-demo.git@v1.2.3", + } + ], + "release": { + "selected_units": [ + {"repo": "govoplan-demo", "version": "1.2.3"} + ] + }, + } + + @staticmethod + def _wheel( + root: Path, + *, + version: str = "1.2.3", + console_script: bool = False, + value: bytes = b"VALUE = 1\n", + ) -> Path: + root.mkdir(parents=True, exist_ok=True) + wheel = root / f"govoplan_demo-{version}-py3-none-any.whl" + dist_info = f"govoplan_demo-{version}.dist-info" + files: dict[str, bytes] = { + "govoplan_demo.py": value, + f"{dist_info}/METADATA": ( + f"Metadata-Version: 2.1\nName: govoplan-demo\nVersion: {version}\n" + ).encode(), + f"{dist_info}/WHEEL": b"Wheel-Version: 1.0\nRoot-Is-Purelib: true\nTag: py3-none-any\n", + } + if console_script: + files[f"{dist_info}/entry_points.txt"] = ( + b"[console_scripts]\ngovoplan-demo = govoplan_demo:main\n" + ) + record_buffer = io.StringIO() + writer = csv.writer(record_buffer, lineterminator="\n") + for name, encoded in files.items(): + writer.writerow((name, "", len(encoded))) + writer.writerow((f"{dist_info}/RECORD", "", "")) + files[f"{dist_info}/RECORD"] = record_buffer.getvalue().encode() + with zipfile.ZipFile(wheel, "w", compression=zipfile.ZIP_DEFLATED) as archive: + for name, encoded in files.items(): + archive.writestr(name, encoded) + return wheel + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/assessments/capability-fit.py b/tools/assessments/capability-fit.py index dc1bbe4..6673dfd 100755 --- a/tools/assessments/capability-fit.py +++ b/tools/assessments/capability-fit.py @@ -105,6 +105,31 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json", ) + parser.add_argument( + "--installer-receipt", + type=Path, + help="Signed installer receipt binding installed payloads to this catalog.", + ) + parser.add_argument( + "--installer-receipt-schema", + type=Path, + default=META_ROOT / "docs" / "installer-receipt.schema.json", + ) + parser.add_argument( + "--installer-authority-keyring", + type=Path, + help=( + "Independently provisioned, role-scoped trust root for installer " + "receipt signatures." + ), + ) + parser.add_argument( + "--installer-authority-keyring-schema", + type=Path, + default=META_ROOT + / "docs" + / "installer-receipt-authority-keyring.schema.json", + ) parser.add_argument( "--boundary-evidence", type=Path, @@ -165,6 +190,18 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit( "--boundary-evidence and --boundary-authority-keyring are required together" ) + if (args.installer_receipt is None) != ( + args.installer_authority_keyring is None + ): + raise SystemExit( + "--installer-receipt and --installer-authority-keyring are required together" + ) + if args.installer_receipt is not None and ( + args.installed_evidence is None and args.collect_installed_evidence is None + ): + raise SystemExit( + "--installer-receipt requires --installed-evidence or --collect-installed-evidence" + ) if args.boundary_evidence is not None and ( args.installed_evidence is None and args.collect_installed_evidence is None ): @@ -286,6 +323,32 @@ def main(argv: list[str] | None = None) -> int: max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) + installer_receipt = None + installer_receipt_schema = None + installer_authority_keyring = None + installer_authority_keyring_schema = None + if args.installer_receipt is not None: + installer_receipt = read_object( + args.installer_receipt, + label="installer receipt", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + installer_receipt_schema = read_object( + args.installer_receipt_schema, + label="installer receipt schema", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + installer_authority_keyring = read_object( + args.installer_authority_keyring, + label="installer authority keyring", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + installer_authority_keyring_schema = read_object( + args.installer_authority_keyring_schema, + label="installer authority keyring schema", + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + ) + verification_time = parse_datetime(args.verification_time) report = review_capability_fit( assessment=assessment, @@ -298,6 +361,10 @@ def main(argv: list[str] | None = None) -> int: else args.workspace_root.resolve(), installed_evidence=installed_evidence, installed_evidence_schema=installed_evidence_schema, + installer_receipt=installer_receipt, + installer_receipt_schema=installer_receipt_schema, + installer_authority_keyring=installer_authority_keyring, + installer_authority_keyring_schema=installer_authority_keyring_schema, boundary_evidence=boundary_evidence, boundary_evidence_schema=boundary_evidence_schema, boundary_authority_keyring=boundary_authority_keyring, diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index d514854..4296851 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -59,6 +59,10 @@ def review_capability_fit( workspace_root: Path | None = None, installed_evidence: dict[str, Any] | None = None, installed_evidence_schema: dict[str, Any] | None = None, + installer_receipt: dict[str, Any] | None = None, + installer_receipt_schema: dict[str, Any] | None = None, + installer_authority_keyring: dict[str, Any] | None = None, + installer_authority_keyring_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, @@ -377,6 +381,24 @@ def review_capability_fit( verification_time=evidence_verification_time, verification_mode=evidence_verification_mode, catalog_trusted=catalog_dependency_trusted, + catalog_artifacts=( + catalog.get("release", {}).get("artifacts") + if isinstance(catalog.get("release"), dict) + and "artifacts" in catalog.get("release", {}) + else None + ), + catalog_sha256=canonical_hash(catalog), + catalog_channel=_text(catalog.get("channel")), + catalog_sequence=_integer(catalog.get("sequence")), + installer_receipt=installer_receipt, + installer_receipt_schema=installer_receipt_schema, + installer_authority_keyring=installer_authority_keyring, + installer_authority_keyring_schema=installer_authority_keyring_schema, + forbidden_installer_public_keys=public_key_material_from_keyrings( + published_keyring, + trusted_keyring, + boundary_authority_keyring or {}, + ), ) findings.extend( Finding(item.severity, item.code, item.message, item.assessment_ids) @@ -398,6 +420,7 @@ def review_capability_fit( forbidden_authority_public_keys=public_key_material_from_keyrings( published_keyring, trusted_keyring, + installer_authority_keyring or {}, ), ) findings.extend( @@ -939,7 +962,7 @@ def build_report( **boundary_scope, } return { - "report_version": "0.4.0", + "report_version": "0.5.0", "status": status, "assessment_id": assessment.get("assessment_id"), "assessment_release": assessment.get("release", {}).get("ref") @@ -977,7 +1000,11 @@ def render_review(report: dict[str, Any]) -> str: ( "Scope: schema, signed release metadata, optional local tag provenance, " + ( - "and locally observed installed-composition consistency evidence; release origin remains a separate proof boundary." + ( + "and locally observed installed-composition evidence independently bound to signed release origin." + if release_origin_checked + else "and locally observed installed-composition consistency evidence; release origin remains a separate proof boundary." + ) if installed_checked else ( "and supplied installed-composition evidence that did not establish local proof." diff --git a/tools/assessments/govoplan_assessment/evidence.py b/tools/assessments/govoplan_assessment/evidence.py index 741462f..56f4263 100644 --- a/tools/assessments/govoplan_assessment/evidence.py +++ b/tools/assessments/govoplan_assessment/evidence.py @@ -25,6 +25,12 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from jsonschema import Draft202012Validator, FormatChecker from jsonschema.exceptions import SchemaError +from govoplan_release.artifact_identity import ( + INSTALLED_PAYLOAD_ALGORITHM, + WHEEL_PAYLOAD_ALGORITHM, + payload_identity_sha256, +) + MAX_GOVOPLAN_DISTRIBUTIONS = 256 MAX_DISTRIBUTION_FILES = 10_000 @@ -243,7 +249,7 @@ def collect_installed_composition( sorted_issues.sort(key=lambda item: (item["package_name"], item["code"])) return { "$schema": "./installed-composition-evidence.schema.json", - "schema_version": "0.3.0", + "schema_version": "0.4.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": _safe_opaque_id( assessment.get("assessment_id"), fallback="invalid-assessment" @@ -277,6 +283,15 @@ def review_installed_composition( verification_time: datetime | None = None, verification_mode: str = "current", catalog_trusted: bool = False, + catalog_artifacts: object = None, + catalog_sha256: str | None = None, + catalog_channel: str | None = None, + catalog_sequence: int | None = None, + installer_receipt: dict[str, Any] | None = None, + installer_receipt_schema: dict[str, Any] | None = None, + installer_authority_keyring: dict[str, Any] | None = None, + installer_authority_keyring_schema: dict[str, Any] | None = None, + forbidden_installer_public_keys: frozenset[bytes] = frozenset(), ) -> InstalledEvidenceReview: if evidence is None: return InstalledEvidenceReview( @@ -403,14 +418,6 @@ def review_installed_composition( observation_trusted = True elif effective_observation_mode == "imported_unsigned": freshness = "unsigned_import" - findings.append( - EvidenceFinding( - "blocker", - "installed_evidence_unsigned_import", - "Imported unsigned installed evidence may be compared but cannot establish installed proof.", - ("assessment.installed_composition",), - ) - ) if effective_verification_mode == "invalid": observation_trusted = False if catalog_trusted is not True: @@ -528,7 +535,7 @@ def review_installed_composition( provenance_declared_match = 0 provenance_mutable = 0 provenance_unanchored = 0 - artifact_valid = binding_valid and not duplicate_packages + payload_consistent = binding_valid and not duplicate_packages for package_name, (component, catalog_entry) in sorted(expected_by_package.items()): module_id = str(component["module_id"]) @@ -537,7 +544,7 @@ def review_installed_composition( record_expected += 1 provenance_expected += 1 if package_name in duplicate_packages: - artifact_valid = False + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -550,7 +557,7 @@ def review_installed_composition( ) continue if artifact is None: - artifact_valid = False + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -567,7 +574,7 @@ def review_installed_composition( 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 + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -606,7 +613,7 @@ def review_installed_composition( and str(expected_manifest.get("manifest_version") or "") != expected_version ) if manifest_mismatch: - artifact_valid = False + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -619,7 +626,7 @@ def review_installed_composition( ) unexpected_modules = sorted(set(module_rows) - {module_id}) if duplicate_module_ids or unexpected_modules: - artifact_valid = False + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -647,7 +654,7 @@ def review_installed_composition( if record_status == "verified" and record_semantics_valid: record_verified += 1 else: - artifact_valid = False + payload_consistent = False _record_change( findings=findings, changes=changes, @@ -676,7 +683,6 @@ def review_installed_composition( str(provenance.get("basis") or "") if isinstance(provenance, dict) else "" ) if provenance_basis != "local-pep610-metadata": - artifact_valid = False provenance_unanchored += 1 _record_change( findings=findings, @@ -703,7 +709,6 @@ def review_installed_composition( if commit_matches and selected_matches: provenance_declared_match += 1 else: - artifact_valid = False _record_change( findings=findings, changes=changes, @@ -716,7 +721,6 @@ def review_installed_composition( ) elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}: provenance_mutable += 1 - artifact_valid = False _record_change( findings=findings, changes=changes, @@ -729,7 +733,6 @@ def review_installed_composition( ) else: provenance_unanchored += 1 - artifact_valid = False _record_change( findings=findings, changes=changes, @@ -743,7 +746,7 @@ def review_installed_composition( expected_packages = set(expected_by_package) for package_name in sorted(set(artifacts_by_package) - expected_packages): - artifact_valid = False + payload_consistent = False matching_module_id = next( ( module_id @@ -788,7 +791,7 @@ def review_installed_composition( continue package_name = str(issue.get("package_name") or "") code = str(issue.get("code") or "") - artifact_valid = False + payload_consistent = False findings.append( EvidenceFinding( "review", @@ -801,18 +804,68 @@ def review_installed_composition( exact_set = ( set(artifacts_by_package) == expected_packages and not duplicate_packages ) - installed_checked = observation_trusted + origin_scope, origin_findings = _review_installed_release_origin( + expected_by_package=expected_by_package, + installed_artifacts=artifacts_by_package, + installed_evidence=evidence, + installed_evidence_sha256=digest, + installed_composition_valid=( + binding_valid + and exact_set + and payload_consistent + and record_verified == record_expected + and record_expected > 0 + ), + observation_trusted=observation_trusted, + catalog_artifacts=catalog_artifacts, + catalog_sha256=catalog_sha256, + catalog_channel=catalog_channel, + catalog_sequence=catalog_sequence, + catalog_trusted=catalog_trusted, + receipt=installer_receipt, + receipt_schema=installer_receipt_schema, + authority_keyring=installer_authority_keyring, + authority_keyring_schema=installer_authority_keyring_schema, + verification_time=now, + forbidden_public_keys=forbidden_installer_public_keys, + ) + findings.extend(origin_findings) + receipt_authenticated = ( + origin_scope.get("valid") is True + and origin_scope.get("signed_installer_receipt_count") == record_expected + and record_expected > 0 + ) + effective_observation_trusted = observation_trusted or receipt_authenticated + if ( + effective_observation_mode == "imported_unsigned" + and not receipt_authenticated + ): + findings.append( + EvidenceFinding( + "blocker", + "installed_evidence_unsigned_import", + "Imported installed evidence requires a valid independent installer receipt before it can establish installed proof.", + ("assessment.installed_composition",), + ) + ) + installed_checked = effective_observation_trusted record_checked = ( - observation_trusted and binding_valid and exact_set and record_expected > 0 + effective_observation_trusted + and binding_valid + and exact_set + and record_expected > 0 ) provenance_checked = ( - observation_trusted and binding_valid and exact_set and provenance_expected > 0 + effective_observation_trusted + and binding_valid + and exact_set + and provenance_expected > 0 ) proof_scope = { "installed_artifacts": { "checked": installed_checked, - "valid": artifact_valid if installed_checked else None, - "comparison_valid": artifact_valid, + "valid": payload_consistent if installed_checked else None, + "comparison_valid": payload_consistent, "evidence_sha256": digest, "expected_distribution_count": len(expected_packages), "observed_distribution_count": len(artifact_rows), @@ -836,16 +889,7 @@ def review_installed_composition( "expected_distribution_count": provenance_expected, "release_origin_bound": False, }, - "installed_release_origin": { - "checked": False, - "valid": None, - "release_origin_bound": False, - "anchored_artifact_digest_count": 0, - "expected_distribution_count": provenance_expected, - "required_evidence": ( - "An independently anchored artifact digest or signed installer receipt that binds each installed payload to the signed release catalog.", - ), - }, + "installed_release_origin": origin_scope, "runtime_activation": { "checked": False, "valid": None, @@ -861,6 +905,7 @@ def review_installed_composition( "evaluated_at": _iso_datetime(now), "verification_mode": effective_verification_mode, "catalog_trusted": catalog_trusted is True, + "receipt_authenticated": receipt_authenticated, "freshness_window_seconds": int(MAX_LOCAL_OBSERVATION_AGE.total_seconds()), }, } @@ -874,6 +919,489 @@ def review_installed_composition( ) +def _review_installed_release_origin( + *, + expected_by_package: Mapping[str, tuple[dict[str, Any], Mapping[str, Any]]], + installed_artifacts: Mapping[str, dict[str, Any]], + installed_evidence: dict[str, Any], + installed_evidence_sha256: str, + installed_composition_valid: bool, + observation_trusted: bool, + catalog_artifacts: object, + catalog_sha256: str | None, + catalog_channel: str | None, + catalog_sequence: int | None, + catalog_trusted: bool, + receipt: dict[str, Any] | None, + receipt_schema: dict[str, Any] | None, + authority_keyring: dict[str, Any] | None, + authority_keyring_schema: dict[str, Any] | None, + verification_time: datetime, + forbidden_public_keys: frozenset[bytes], +) -> tuple[dict[str, Any], tuple[EvidenceFinding, ...]]: + findings: list[EvidenceFinding] = [] + expected_count = len(expected_by_package) + scope: dict[str, Any] = { + "checked": False, + "valid": None, + "release_origin_bound": False, + "anchored_artifact_digest_count": 0, + "signed_installer_receipt_count": 0, + "expected_distribution_count": expected_count, + "catalog_sha256": catalog_sha256, + "installer_receipt_sha256": None, + "required_evidence": ( + "A signed catalog identity computed from each built immutable artifact, or a role-scoped installer receipt signed by an independently provisioned authority.", + ), + } + manifest_supplied = catalog_artifacts is not None + receipt_supplied = receipt is not None + if not manifest_supplied and not receipt_supplied: + return scope, () + scope["checked"] = True + scope["valid"] = False + if not installed_composition_valid or catalog_trusted is not True: + findings.append( + EvidenceFinding( + "blocker", + "installed_origin_dependency_untrusted", + "Installed release-origin proof requires a trusted catalog and valid fresh installed-composition evidence.", + ("assessment.release", "assessment.installed_composition"), + ) + ) + return scope, tuple(findings) + + catalog_by_package, catalog_findings = _catalog_artifact_identities( + catalog_artifacts + ) + findings.extend(catalog_findings) + if catalog_findings: + return scope, tuple(findings) + + direct_bound: set[str] = set() + receipt_required: set[str] = set() + for package_name, (component, catalog_entry) in sorted(expected_by_package.items()): + expected_version = str(catalog_entry.get("version") or "") + identity = catalog_by_package.get(package_name) + if identity is None or identity.get("package_version") != expected_version: + findings.append( + EvidenceFinding( + "review", + "installed_origin_catalog_artifact_missing", + f"The signed catalog has no built-artifact identity for {package_name!r} at the selected version.", + (f"composition.{component['module_id']}",), + ) + ) + continue + if identity.get("requires_installer_receipt") is True: + receipt_required.add(package_name) + continue + if not observation_trusted: + continue + installed = installed_artifacts.get(package_name) + record = installed.get("record_integrity") if isinstance(installed, dict) else None + observed = ( + record.get("artifact_payload_identity") if isinstance(record, dict) else None + ) + expected = identity.get("installed_payload") + if observed == expected: + direct_bound.add(package_name) + else: + findings.append( + EvidenceFinding( + "blocker", + "installed_origin_artifact_payload_mismatch", + f"Installed payload for {package_name!r} does not match the identity independently computed from the signed catalog artifact.", + (f"composition.{component['module_id']}",), + ) + ) + + receipt_bound: set[str] = set() + if receipt_supplied: + receipt_bound, receipt_findings, receipt_digest = _verify_installer_receipt( + expected_by_package=expected_by_package, + installed_artifacts=installed_artifacts, + installed_evidence=installed_evidence, + installed_evidence_sha256=installed_evidence_sha256, + catalog_by_package=catalog_by_package, + catalog_sha256=catalog_sha256, + catalog_channel=catalog_channel, + catalog_sequence=catalog_sequence, + receipt=receipt, + receipt_schema=receipt_schema, + authority_keyring=authority_keyring, + authority_keyring_schema=authority_keyring_schema, + verification_time=verification_time, + forbidden_public_keys=forbidden_public_keys, + ) + findings.extend(receipt_findings) + scope["installer_receipt_sha256"] = receipt_digest + elif receipt_required: + findings.append( + EvidenceFinding( + "review", + "installed_origin_receipt_required", + "At least one selected wheel produces installer-transformed files, so its signed installer receipt is required for complete origin proof.", + ("assessment.installed_composition",), + ) + ) + + bound = direct_bound | receipt_bound + scope["anchored_artifact_digest_count"] = len(direct_bound) + scope["signed_installer_receipt_count"] = len(receipt_bound) + valid = len(bound) == expected_count and not any( + finding.severity == "blocker" for finding in findings + ) + scope["valid"] = valid + scope["release_origin_bound"] = valid + return scope, tuple(findings) + + +def _catalog_artifact_identities( + value: object, +) -> tuple[dict[str, dict[str, Any]], tuple[EvidenceFinding, ...]]: + if value is None: + return {}, () + if not isinstance(value, list) or len(value) > MAX_GOVOPLAN_DISTRIBUTIONS: + return {}, ( + EvidenceFinding( + "blocker", + "catalog_artifact_manifest_invalid", + "Signed catalog release artifacts must be a bounded list.", + ("assessment.release",), + ), + ) + result: dict[str, dict[str, Any]] = {} + invalid = False + for item in value: + if not isinstance(item, dict) or set(item) != { + "artifact_kind", + "package_name", + "package_version", + "archive_sha256", + "archive_size", + "installed_payload", + "requires_installer_receipt", + }: + invalid = True + continue + package_name = item.get("package_name") + version = item.get("package_version") + archive_size = item.get("archive_size") + payload = item.get("installed_payload") + if ( + item.get("artifact_kind") != "python-wheel" + or not isinstance(package_name, str) + or PACKAGE_PATTERN.fullmatch(package_name) is None + or not isinstance(version, str) + or VERSION_PATTERN.fullmatch(version) is None + or not isinstance(item.get("archive_sha256"), str) + or SHA256_PATTERN.fullmatch(str(item["archive_sha256"])) is None + or not isinstance(archive_size, int) + or isinstance(archive_size, bool) + or not 0 < archive_size <= MAX_HASHED_DISTRIBUTION_BYTES + or not isinstance(item.get("requires_installer_receipt"), bool) + or not _payload_identity_semantics_valid( + payload, + algorithm=WHEEL_PAYLOAD_ALGORITHM, + maximum_files=MAX_DISTRIBUTION_FILES, + ) + or package_name in result + ): + invalid = True + continue + result[package_name] = item + if invalid: + return {}, ( + EvidenceFinding( + "blocker", + "catalog_artifact_manifest_invalid", + "Signed catalog release artifacts contain malformed or duplicate package identities.", + ("assessment.release",), + ), + ) + return result, () + + +def _verify_installer_receipt( + *, + expected_by_package: Mapping[str, tuple[dict[str, Any], Mapping[str, Any]]], + installed_artifacts: Mapping[str, dict[str, Any]], + installed_evidence: dict[str, Any], + installed_evidence_sha256: str, + catalog_by_package: Mapping[str, dict[str, Any]], + catalog_sha256: str | None, + catalog_channel: str | None, + catalog_sequence: int | None, + receipt: dict[str, Any], + receipt_schema: dict[str, Any] | None, + authority_keyring: dict[str, Any] | None, + authority_keyring_schema: dict[str, Any] | None, + verification_time: datetime, + forbidden_public_keys: frozenset[bytes], +) -> tuple[set[str], tuple[EvidenceFinding, ...], str]: + receipt_digest = canonical_sha256(receipt) + findings: list[EvidenceFinding] = [] + if receipt_schema is None or authority_keyring_schema is None: + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_schema_missing", + "Installer receipt proof requires both bounded validation schemas.", + ("assessment.installed_composition",), + ) + ) + return set(), tuple(findings), receipt_digest + receipt_errors = validate_payload(payload=receipt, schema=receipt_schema) + keyring_errors = ( + validate_payload(payload=authority_keyring, schema=authority_keyring_schema) + if isinstance(authority_keyring, dict) + else ("$: an independently provisioned installer authority keyring is required",) + ) + findings.extend( + EvidenceFinding( + "blocker", + "installer_receipt_schema", + message, + ("assessment.installed_composition",), + ) + for message in receipt_errors + ) + findings.extend( + EvidenceFinding( + "blocker", + "installer_authority_keyring_schema", + message, + ("assessment.installed_composition",), + ) + for message in keyring_errors + ) + if receipt_errors or keyring_errors or not isinstance(authority_keyring, dict): + return set(), tuple(findings), receipt_digest + + expected_release = installed_evidence.get("assessment_release") + expected_assessment = installed_evidence.get("assessment_id") + catalog_binding = receipt.get("catalog") + binding_valid = True + if ( + receipt.get("assessment_id") != expected_assessment + or receipt.get("assessment_release") != expected_release + or receipt.get("installed_evidence_sha256") != installed_evidence_sha256 + ): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_evidence_mismatch", + "Installer receipt is not bound to the supplied assessment and installed-evidence digest.", + ("assessment.installed_composition",), + ) + ) + if not isinstance(catalog_binding, dict) or ( + catalog_binding.get("sha256") != catalog_sha256 + or catalog_binding.get("channel") != catalog_channel + or catalog_binding.get("sequence") != catalog_sequence + ): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_catalog_mismatch", + "Installer receipt is not bound to the supplied signed catalog identity.", + ("assessment.release", "assessment.installed_composition"), + ) + ) + + issued_at = _datetime(receipt.get("issued_at")) + collected_at = _datetime(installed_evidence.get("collected_at")) + if ( + issued_at is None + or collected_at is None + or issued_at < collected_at + or issued_at - collected_at > MAX_LOCAL_OBSERVATION_AGE + or issued_at > verification_time + MAX_LOCAL_OBSERVATION_FUTURE_SKEW + or verification_time - issued_at > MAX_LOCAL_OBSERVATION_AGE + ): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_time_invalid", + "Installer receipt issuance must follow the observation within five minutes and remain fresh at the selected verification time.", + ("assessment.installed_composition",), + ) + ) + + receipt_rows = [ + item for item in receipt.get("artifacts", []) if isinstance(item, dict) + ] + rows_by_package: dict[str, dict[str, Any]] = {} + duplicate_packages: set[str] = set() + for row in receipt_rows: + package_name = str(row.get("package_name") or "") + if package_name in rows_by_package: + duplicate_packages.add(package_name) + else: + rows_by_package[package_name] = row + expected_packages = set(expected_by_package) + if duplicate_packages or set(rows_by_package) != expected_packages: + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_artifact_set_mismatch", + "Installer receipt must bind exactly one row for every expected installed distribution.", + ("assessment.installed_composition",), + ) + ) + bound: set[str] = set() + for package_name, (_, catalog_entry) in sorted(expected_by_package.items()): + row = rows_by_package.get(package_name) + installed = installed_artifacts.get(package_name) + record = installed.get("record_integrity") if isinstance(installed, dict) else None + observed_payload = ( + record.get("installed_payload_identity") if isinstance(record, dict) else None + ) + catalog_identity = catalog_by_package.get(package_name) + if ( + row is None + or catalog_identity is None + or row.get("package_version") != catalog_entry.get("version") + or row.get("catalog_archive_sha256") + != catalog_identity.get("archive_sha256") + or row.get("installed_payload") != observed_payload + ): + binding_valid = False + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_artifact_mismatch", + f"Installer receipt does not bind the observed {package_name!r} payload to its signed catalog artifact.", + ("assessment.installed_composition",), + ) + ) + else: + bound.add(package_name) + + keys, key_findings = _installer_receipt_keys( + authority_keyring=authority_keyring, + issued_at=issued_at, + forbidden_public_keys=forbidden_public_keys, + ) + findings.extend(key_findings) + signature_payload = dict(receipt) + signature_payload.pop("signatures", None) + signature_bytes = canonical_bytes(signature_payload) + signature_valid = False + seen_ids: set[str] = set() + duplicate_ids = False + for signature in receipt.get("signatures", []): + if not isinstance(signature, dict): + continue + key_id = str(signature.get("key_id") or "") + if key_id in seen_ids: + duplicate_ids = True + continue + seen_ids.add(key_id) + public_key = keys.get(key_id) + if public_key is None or signature.get("algorithm") != "ed25519": + continue + try: + encoded = base64.b64decode(str(signature.get("value") or ""), validate=True) + Ed25519PublicKey.from_public_bytes(public_key).verify( + encoded, signature_bytes + ) + signature_valid = True + except (ValueError, InvalidSignature): + continue + if duplicate_ids or not signature_valid: + findings.append( + EvidenceFinding( + "blocker", + "installer_receipt_signature_untrusted", + "Installer receipt has no unique valid signature from an authorized independent installer key.", + ("assessment.installed_composition",), + ) + ) + if not binding_valid or not signature_valid or key_findings: + return set(), tuple(findings), receipt_digest + return bound, tuple(findings), receipt_digest + + +def _installer_receipt_keys( + *, + authority_keyring: dict[str, Any], + issued_at: datetime | None, + forbidden_public_keys: frozenset[bytes], +) -> tuple[dict[str, bytes], tuple[EvidenceFinding, ...]]: + findings: list[EvidenceFinding] = [] + result: dict[str, bytes] = {} + raw_keys = authority_keyring.get("keys") + if not isinstance(raw_keys, list) or issued_at is None: + return {}, ( + EvidenceFinding( + "blocker", + "installer_authority_untrusted", + "Installer receipt authority keyring or issuance time is invalid.", + ("assessment.installed_composition",), + ), + ) + seen_ids: set[str] = set() + seen_material: set[bytes] = set() + malformed = False + for item in raw_keys: + if not isinstance(item, dict): + malformed = True + continue + key_id = str(item.get("key_id") or "") + try: + public_key = base64.b64decode(str(item.get("public_key") or ""), validate=True) + Ed25519PublicKey.from_public_bytes(public_key) + except ValueError: + malformed = True + continue + not_before = _datetime(item.get("not_before")) + not_after = _datetime(item.get("not_after")) + interval_valid = not ( + not_before is not None + and not_after is not None + and not_before >= not_after + ) + if ( + OPAQUE_ID_PATTERN.fullmatch(key_id) is None + or key_id in seen_ids + or public_key in seen_material + or public_key in forbidden_public_keys + or item.get("allowed_scopes") != ["installed_release_origin"] + or not interval_valid + ): + malformed = True + continue + seen_ids.add(key_id) + seen_material.add(public_key) + status = item.get("status") + active_at_issue = ( + (not_before is None or issued_at >= not_before) + and (not_after is None or issued_at < not_after) + ) + historically_retired = status == "retired" and not_after is not None + if active_at_issue and ( + status in {"active", "next"} or historically_retired + ): + result[key_id] = public_key + if malformed or not result: + findings.append( + EvidenceFinding( + "blocker", + "installer_authority_untrusted", + "Installer authority keyring is malformed, reuses another trust-domain key, or has no role-scoped key valid at receipt issuance.", + ("assessment.installed_composition",), + ) + ) + return {}, tuple(findings) + return result, () + + def review_boundary_evidence( *, assessment: dict[str, Any], @@ -1606,13 +2134,17 @@ def _record_integrity( "missing_file_count": 0, "mismatched_file_count": 0, } + identities = { + "artifact_payload_identity": None, + "installed_payload_identity": None, + } try: distribution_root = Path(distribution.locate_file("")).resolve() installation_root = Path(sys.prefix).resolve() except Exception: - return {"status": "unavailable", **counts} + return {"status": "unavailable", **counts, **identities} if not _trusted_distribution_root(distribution_root): - return {"status": "unavailable", **counts} + return {"status": "unavailable", **counts, **identities} entries, malformed_rows, record_status = _read_record_entries( distribution=distribution, distribution_root=distribution_root, @@ -1620,13 +2152,13 @@ def _record_integrity( ) if record_status == "limit-exceeded": budget.remaining_files = 0 - return {"status": "limit-exceeded", **counts} + return {"status": "limit-exceeded", **counts, **identities} if record_status != "ok": counts["unverifiable_file_count"] = malformed_rows - return {"status": "unavailable", **counts} + return {"status": "unavailable", **counts, **identities} record_file_count = len(entries) + malformed_rows if record_file_count == 0: - return {"status": "unavailable", **counts} + return {"status": "unavailable", **counts, **identities} budget.remaining_files -= record_file_count counts["unverifiable_file_count"] = malformed_rows declared_scripts = _declared_distribution_scripts(distribution) @@ -1666,6 +2198,8 @@ def _record_integrity( } total_hashed_bytes = 0 limit_exceeded = False + artifact_payload_rows: list[dict[str, object]] = [] + installed_payload_rows: list[dict[str, object]] = [] for entry, resolved_path in resolved_entries: if resolved_path is None: counts["unverifiable_file_count"] += 1 @@ -1731,6 +2265,31 @@ def _record_integrity( continue if hmac.compare_digest(actual, expected): counts["hashed_file_count"] += 1 + installed_path = _installed_payload_path( + resolved_path, + distribution_root=distribution_root, + installation_root=installation_root, + declared_scripts=declared_scripts, + ) + if installed_path is None: + counts["unverifiable_file_count"] += 1 + continue + row = { + "path": installed_path, + "sha256": actual.hex(), + "size": file_stat.st_size, + } + installed_payload_rows.append(row) + artifact_path = _artifact_payload_path( + resolved_path, + distribution_root=distribution_root, + installation_root=installation_root, + declared_scripts=declared_scripts, + ) + if artifact_path is not None and not _installer_generated_metadata( + resolved_path, distribution_root=distribution_root + ): + artifact_payload_rows.append({**row, "path": artifact_path}) else: counts["mismatched_file_count"] += 1 if limit_exceeded: @@ -1743,7 +2302,85 @@ def _record_integrity( status = "partial" else: status = "verified" - return {"status": status, **counts} + if status == "verified": + installed_identity = _payload_identity( + algorithm=INSTALLED_PAYLOAD_ALGORITHM, + rows=installed_payload_rows, + ) + artifact_identity = _payload_identity( + algorithm=WHEEL_PAYLOAD_ALGORITHM, + rows=artifact_payload_rows, + ) + if installed_identity is None or artifact_identity is None: + status = "partial" + counts["unverifiable_file_count"] += 1 + else: + identities["installed_payload_identity"] = installed_identity + identities["artifact_payload_identity"] = artifact_identity + return {"status": status, **counts, **identities} + + +def _payload_identity( + *, algorithm: str, rows: list[dict[str, object]] +) -> dict[str, object] | None: + ordered = sorted(rows, key=lambda item: str(item["path"])) + paths = [str(item["path"]) for item in ordered] + if not ordered or len(set(paths)) != len(paths): + return None + return { + "algorithm": algorithm, + "sha256": payload_identity_sha256(algorithm=algorithm, rows=ordered), + "file_count": len(ordered), + } + + +def _installed_payload_path( + path: Path, + *, + distribution_root: Path, + installation_root: Path, + declared_scripts: frozenset[str], +) -> str | None: + if _is_relative_to(path, distribution_root): + return path.relative_to(distribution_root).as_posix() + for script_root in ( + (installation_root / "bin").resolve(), + (installation_root / "Scripts").resolve(), + ): + if path.parent == script_root and path.name in declared_scripts: + return f"@scripts/{path.name}" + if _is_relative_to(path, installation_root): + return f"@data/{path.relative_to(installation_root).as_posix()}" + return None + + +def _artifact_payload_path( + path: Path, + *, + distribution_root: Path, + installation_root: Path, + declared_scripts: frozenset[str], +) -> str | None: + installed = _installed_payload_path( + path, + distribution_root=distribution_root, + installation_root=installation_root, + declared_scripts=declared_scripts, + ) + return None if installed is None or installed.startswith("@scripts/") else installed + + +def _installer_generated_metadata( + path: Path, *, distribution_root: Path +) -> bool: + if not _is_relative_to(path, distribution_root): + return False + relative = path.relative_to(distribution_root) + return ( + len(relative.parts) >= 2 + and relative.parent.name.endswith(".dist-info") + and relative.name in {"direct_url.json", "INSTALLER", "REQUESTED"} + ) def _resolve_record_path( @@ -1867,12 +2504,25 @@ def _record_integrity_semantics_valid(record: Mapping[str, Any]) -> bool: missing = counts["missing_file_count"] mismatched = counts["mismatched_file_count"] if status == "verified": + artifact_identity = record.get("artifact_payload_identity") + installed_identity = record.get("installed_payload_identity") return ( hashed > 0 and counts["permitted_unhashed_file_count"] > 0 and unverifiable == 0 and missing == 0 and mismatched == 0 + and _payload_identity_semantics_valid( + artifact_identity, + algorithm=WHEEL_PAYLOAD_ALGORITHM, + maximum_files=hashed, + ) + and _payload_identity_semantics_valid( + installed_identity, + algorithm=INSTALLED_PAYLOAD_ALGORITHM, + maximum_files=hashed, + exact_files=hashed, + ) ) if status == "partial": return hashed > 0 and unverifiable > 0 and missing == 0 and mismatched == 0 @@ -1885,6 +2535,31 @@ def _record_integrity_semantics_valid(record: Mapping[str, Any]) -> bool: return False +def _payload_identity_semantics_valid( + value: object, + *, + algorithm: str, + maximum_files: int, + exact_files: int | None = None, +) -> bool: + if not isinstance(value, Mapping) or set(value) != { + "algorithm", + "sha256", + "file_count", + }: + return False + file_count = value.get("file_count") + return ( + value.get("algorithm") == algorithm + and isinstance(value.get("sha256"), str) + and SHA256_PATTERN.fullmatch(str(value["sha256"])) is not None + and isinstance(file_count, int) + and not isinstance(file_count, bool) + and 0 < file_count <= maximum_files + and (exact_files is None or file_count == exact_files) + ) + + def _permitted_unhashed_record_path( path: Path, *, diff --git a/tools/assessments/govoplan_assessment/installer_receipt.py b/tools/assessments/govoplan_assessment/installer_receipt.py new file mode 100644 index 0000000..c4ba9b2 --- /dev/null +++ b/tools/assessments/govoplan_assessment/installer_receipt.py @@ -0,0 +1,204 @@ +"""Issuance of role-scoped receipts from same-process installed observations.""" + +from __future__ import annotations + +import base64 +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Mapping + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .capability_fit import canonical_hash +from .evidence import ( + MAX_LOCAL_OBSERVATION_AGE, + OPAQUE_ID_PATTERN, + _catalog_artifact_identities, + _record_integrity_semantics_valid, + canonical_bytes, + canonical_sha256, + normalize_package_name, +) +from govoplan_release.artifact_identity import inspect_python_wheel + + +def issue_installer_receipt( + *, + assessment: Mapping[str, Any], + catalog: Mapping[str, Any], + installed_evidence: dict[str, Any], + receipt_id: str, + key_id: str, + private_key: Ed25519PrivateKey, + consumed_artifacts: Mapping[str, Path | str] | None = None, + issued_at: datetime | None = None, + same_process_observation: bool = False, +) -> dict[str, Any]: + """Sign a receipt only for a fresh observation collected by this process. + + File-based evidence is intentionally not an admitted issuance input. The CLI + collector calls this function directly with its in-memory observation. + """ + + if same_process_observation is not True: + raise ValueError("installer receipts require a same-process observation") + if OPAQUE_ID_PATTERN.fullmatch(receipt_id) is None: + raise ValueError("receipt ID must be a bounded opaque ID") + if OPAQUE_ID_PATTERN.fullmatch(key_id) is None: + raise ValueError("installer signing key ID must be a bounded opaque ID") + if not isinstance(consumed_artifacts, Mapping) or not consumed_artifacts: + raise ValueError("installer receipt issuance requires exact consumed wheels") + release = catalog.get("release") + raw_artifacts = release.get("artifacts") if isinstance(release, Mapping) else None + catalog_artifacts, artifact_findings = _catalog_artifact_identities(raw_artifacts) + if artifact_findings: + raise ValueError("signed catalog artifact manifest is invalid") + evidence_rows = installed_evidence.get("artifacts") + if not isinstance(evidence_rows, list) or not evidence_rows: + raise ValueError("installed evidence has no artifact observations") + expected_packages = _expected_packages(assessment=assessment, catalog=catalog) + receipt_rows: list[dict[str, Any]] = [] + seen: set[str] = set() + for artifact in evidence_rows: + if not isinstance(artifact, dict): + raise ValueError("installed evidence contains a malformed artifact") + package_name = normalize_package_name(str(artifact.get("package_name") or "")) + package_version = str(artifact.get("package_version") or "") + if package_name in seen: + raise ValueError("installed evidence contains duplicate packages") + seen.add(package_name) + catalog_artifact = catalog_artifacts.get(package_name) + record = artifact.get("record_integrity") + installed_payload = ( + record.get("installed_payload_identity") + if isinstance(record, Mapping) + else None + ) + if ( + catalog_artifact is None + or catalog_artifact.get("package_version") != package_version + or not isinstance(record, Mapping) + or record.get("status") != "verified" + or not _record_integrity_semantics_valid(record) + or not isinstance(installed_payload, Mapping) + ): + raise ValueError( + "installed payload is not a verified match for a catalog artifact" + ) + artifact_path = consumed_artifacts.get(package_name) + if artifact_path is None: + raise ValueError("every installed package requires its exact consumed wheel") + consumed = inspect_python_wheel(artifact_path) + if ( + consumed.package_name != package_name + or consumed.package_version != package_version + or consumed.archive_sha256 != catalog_artifact.get("archive_sha256") + or consumed.catalog_payload().get("installed_payload") + != catalog_artifact.get("installed_payload") + or record.get("artifact_payload_identity") + != catalog_artifact.get("installed_payload") + ): + raise ValueError( + "consumed wheel or observed wheel payload differs from the signed catalog identity" + ) + receipt_rows.append( + { + "package_name": package_name, + "package_version": package_version, + "catalog_archive_sha256": catalog_artifact["archive_sha256"], + "installed_payload": dict(installed_payload), + } + ) + if seen != expected_packages: + raise ValueError( + "installed evidence must contain exactly the enabled assessed catalog packages" + ) + if set(consumed_artifacts) != expected_packages: + raise ValueError("consumed wheel set must exactly match enabled packages") + receipt_rows.sort(key=lambda item: (item["package_name"], item["package_version"])) + observed_at = issued_at or datetime.now(tz=UTC) + if observed_at.tzinfo is None: + raise ValueError("installer receipt issuance time must include a timezone") + collected_at = _datetime(installed_evidence.get("collected_at")) + if ( + collected_at is None + or observed_at < collected_at + or observed_at - collected_at > MAX_LOCAL_OBSERVATION_AGE + ): + raise ValueError( + "installer receipt issuance must follow live collection within five minutes" + ) + assessment_release = assessment.get("release") + assessment_release_ref = ( + assessment_release.get("ref") + if isinstance(assessment_release, Mapping) + else None + ) + receipt: dict[str, Any] = { + "$schema": "./installer-receipt.schema.json", + "schema_version": "0.1.0", + "evidence_kind": "govoplan.installer-receipt", + "receipt_id": receipt_id, + "assessment_id": assessment.get("assessment_id"), + "assessment_release": assessment_release_ref, + "installed_evidence_sha256": canonical_sha256(installed_evidence), + "catalog": { + "channel": catalog.get("channel"), + "sequence": catalog.get("sequence"), + "sha256": canonical_hash(catalog), + }, + "issued_at": observed_at.astimezone(UTC).isoformat().replace("+00:00", "Z"), + "artifacts": receipt_rows, + } + receipt["signatures"] = [ + { + "algorithm": "ed25519", + "key_id": key_id, + "value": base64.b64encode( + private_key.sign(canonical_bytes(receipt)) + ).decode("ascii"), + } + ] + return receipt + + +def _expected_packages( + *, assessment: Mapping[str, Any], catalog: Mapping[str, Any] +) -> set[str]: + entries: dict[str, Mapping[str, Any]] = {} + core = catalog.get("core_release") + if isinstance(core, Mapping): + entries["core"] = core + modules = catalog.get("modules") + if isinstance(modules, list): + for entry in modules: + if isinstance(entry, Mapping) and isinstance(entry.get("module_id"), str): + entries[str(entry["module_id"])] = entry + expected: set[str] = set() + composition = assessment.get("composition") + if not isinstance(composition, list): + raise ValueError("assessment composition is unavailable") + for component in composition: + if not isinstance(component, Mapping) or component.get("enabled") is not True: + continue + entry = entries.get(str(component.get("module_id") or "")) + package = entry.get("python_package") if isinstance(entry, Mapping) else None + if not isinstance(package, str): + raise ValueError("enabled assessment component has no catalog package") + normalized = normalize_package_name(package) + if normalized in expected: + raise ValueError("enabled assessment packages are ambiguous") + expected.add(normalized) + if not expected: + raise ValueError("assessment has no enabled catalog packages") + return expected + + +def _datetime(value: object) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed.astimezone(UTC) if parsed.tzinfo is not None else None diff --git a/tools/assessments/installer-receipt.py b/tools/assessments/installer-receipt.py new file mode 100644 index 0000000..dd510f2 --- /dev/null +++ b/tools/assessments/installer-receipt.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Collect live installed payloads and issue an independently signed receipt.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + + +META_ROOT = Path(__file__).resolve().parents[2] +for tools_root in (META_ROOT / "tools" / "assessments", META_ROOT / "tools" / "release"): + if str(tools_root) not in sys.path: + sys.path.insert(0, str(tools_root)) + +from cryptography.hazmat.primitives import serialization # noqa: E402 +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( # noqa: E402 + Ed25519PrivateKey, +) + +from govoplan_assessment import collect_installed_composition # noqa: E402 +from govoplan_assessment.atomic_io import ( # noqa: E402 + AtomicJsonWriteError, + atomic_write_json, +) +from govoplan_assessment.capability_fit import ( # noqa: E402 + canonical_hash, + review_capability_fit, +) +from govoplan_assessment.evidence import validate_payload # noqa: E402 +from govoplan_assessment.installer_receipt import ( # noqa: E402 + issue_installer_receipt, +) + + +MAX_INPUT_BYTES = 16 * 1024 * 1024 +MAX_OUTPUT_BYTES = 4 * 1024 * 1024 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--assessment", type=Path, required=True) + parser.add_argument("--assessment-schema", type=Path, default=META_ROOT / "docs" / "capability-fit.schema.json") + parser.add_argument("--catalog", type=Path, required=True) + parser.add_argument("--keyring", type=Path, required=True) + parser.add_argument("--trusted-keyring", type=Path, required=True) + parser.add_argument("--receipt-id", required=True) + parser.add_argument("--signing-key", required=True, metavar="KEY_ID=/PATH/PRIVATE.pem") + parser.add_argument( + "--python-artifact", + action="append", + default=[], + required=True, + metavar="PACKAGE=/PATH/TO/CONSUMED.whl", + help="Exact wheel consumed by this installer; repeat for every package.", + ) + parser.add_argument("--evidence-output", type=Path, required=True) + parser.add_argument("--receipt-output", type=Path, required=True) + parser.add_argument("--installed-evidence-schema", type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json") + parser.add_argument("--receipt-schema", type=Path, default=META_ROOT / "docs" / "installer-receipt.schema.json") + args = parser.parse_args(argv) + if args.evidence_output.resolve() == args.receipt_output.resolve(): + parser.error("evidence and receipt output paths must differ") + + assessment = read_object(args.assessment, label="assessment") + assessment_schema = read_object(args.assessment_schema, label="assessment schema") + catalog = read_object(args.catalog, label="catalog") + keyring = read_object(args.keyring, label="published keyring") + trusted_keyring = read_object(args.trusted_keyring, label="trusted keyring") + evidence_schema = read_object(args.installed_evidence_schema, label="installed evidence schema") + receipt_schema = read_object(args.receipt_schema, label="installer receipt schema") + release = catalog.get("release") + if not isinstance(release, dict) or release.get("keyring_sha256") != canonical_hash(keyring): + parser.error("catalog does not pin the supplied published keyring") + + evidence = collect_installed_composition(assessment=assessment) + if validate_payload(payload=evidence, schema=evidence_schema): + parser.error("live installed observation does not satisfy its bounded schema") + report = review_capability_fit( + assessment=assessment, + schema=assessment_schema, + catalog=catalog, + published_keyring=keyring, + trusted_keyring=trusted_keyring, + workspace_root=None, + installed_evidence=evidence, + installed_evidence_schema=evidence_schema, + installed_evidence_mode="direct_local", + ) + required_scopes = ( + "catalog_signature_and_keyring", + "catalog_signature_and_trusted_keyring", + "published_keyring_hash", + "release_metadata", + "installed_artifacts", + "installed_record_integrity", + ) + if any(report["proof_scope"].get(scope, {}).get("valid") is not True for scope in required_scopes): + parser.error("trusted catalog and live installed-composition checks must pass before receipt issuance") + + key_id, private_key = read_signing_key(args.signing_key) + consumed_artifacts = parse_package_paths(tuple(args.python_artifact)) + receipt = issue_installer_receipt( + assessment=assessment, + catalog=catalog, + installed_evidence=evidence, + receipt_id=args.receipt_id, + key_id=key_id, + private_key=private_key, + consumed_artifacts=consumed_artifacts, + same_process_observation=True, + ) + if validate_payload(payload=receipt, schema=receipt_schema): + parser.error("issued installer receipt does not satisfy its bounded schema") + write_object(args.evidence_output, evidence, label="installed evidence") + write_object(args.receipt_output, receipt, label="installer receipt") + print(json.dumps({"receipt_id": args.receipt_id, "evidence_output": str(args.evidence_output), "receipt_output": str(args.receipt_output)}, sort_keys=True)) + return 0 + + +def read_object(path: Path, *, label: str) -> dict[str, object]: + try: + if path.stat().st_size > MAX_INPUT_BYTES: + raise ValueError("input exceeds size limit") + encoded = path.read_bytes() + payload = json.loads(encoded.decode("utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: + raise SystemExit(f"Could not read {label}") from exc + if not isinstance(payload, dict): + raise SystemExit(f"{label.capitalize()} must be a JSON object") + return payload + + +def read_signing_key(value: str) -> tuple[str, Ed25519PrivateKey]: + key_id, separator, path_text = value.partition("=") + if not separator or not key_id or not path_text: + raise SystemExit("--signing-key must use KEY_ID=/path/to/private.pem") + try: + key = serialization.load_pem_private_key(Path(path_text).expanduser().read_bytes(), password=None) + except (OSError, ValueError) as exc: + raise SystemExit("Could not read installer signing key") from exc + if not isinstance(key, Ed25519PrivateKey): + raise SystemExit("Installer signing key must be Ed25519") + return key_id, key + + +def parse_package_paths(values: tuple[str, ...]) -> dict[str, Path]: + result: dict[str, Path] = {} + for value in values: + package, separator, path_text = value.partition("=") + package = package.strip() + path_text = path_text.strip() + if not separator or not package or not path_text or package in result: + raise SystemExit( + "--python-artifact must uniquely use PACKAGE=/path/to/consumed.whl" + ) + result[package] = Path(path_text).expanduser() + return result + + +def write_object(path: Path, payload: dict[str, object], *, label: str) -> None: + try: + atomic_write_json(path, payload, max_bytes=MAX_OUTPUT_BYTES) + except AtomicJsonWriteError as exc: + raise SystemExit(f"Could not securely write {label}") from exc + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/release/govoplan_release/selective_catalog.py b/tools/release/govoplan_release/selective_catalog.py index 4f3896a..1e26aca 100644 --- a/tools/release/govoplan_release/selective_catalog.py +++ b/tools/release/govoplan_release/selective_catalog.py @@ -5,6 +5,7 @@ from __future__ import annotations import base64 from datetime import UTC, datetime, timedelta import json +import os from pathlib import Path import re from typing import Any @@ -15,6 +16,11 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from govoplan_core.core.module_package_catalog import validate_module_package_catalog from .catalog import DEFAULT_PUBLIC_BASE_URL, canonical_hash, fetch_json +from .artifact_identity import inspect_python_wheel +from .candidate_artifact import ( + harden_private_candidate_tree, + validate_release_channel, +) from .catalog_entry_synthesis import ( synthesize_repository_catalog_entries, validate_initial_entry_closure, @@ -52,7 +58,9 @@ def build_selective_catalog_candidate( sequence: int | None = None, check_public: bool = True, write_summary: bool = True, + python_artifacts: dict[str, Path | str] | None = None, ) -> SelectiveCatalogCandidate: + channel = validate_release_channel(channel) if not repo_versions: raise ValueError("At least one repo version must be provided.") parsed_keys = tuple(parse_signing_key(value) for value in signing_keys) @@ -107,6 +115,13 @@ def build_selective_catalog_candidate( repository_base=repository_base.rstrip("/"), workspace=workspace, ) + changes.extend( + apply_python_artifact_identities( + candidate, + repo_versions=repo_versions, + python_artifacts=python_artifacts or {}, + ) + ) enforce_complete_catalog_source_provenance( payload=candidate, workspace=workspace, @@ -128,10 +143,15 @@ def build_selective_catalog_candidate( candidate.pop("signatures", None) candidate["signatures"] = [signature(candidate, key_id=key_id, private_key=private_key) for key_id, private_key in parsed_keys] - catalog_path.parent.mkdir(parents=True, exist_ok=True) + output_root.mkdir(mode=0o700, parents=True, exist_ok=True) + harden_private_candidate_tree(output_root) + catalog_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(catalog_path.parent, 0o700, follow_symlinks=False) catalog_path.write_text(json.dumps(candidate, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.chmod(catalog_path, 0o600, follow_symlinks=False) keyring_path.write_text(json.dumps(keyring_payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.chmod(keyring_path, 0o600, follow_symlinks=False) module_directory_files = write_module_directory( catalog_payload=candidate, keyring_payload=keyring_payload, @@ -196,6 +216,7 @@ def build_selective_catalog_candidate( ) if summary_path is not None: summary_path.write_text(json.dumps(dataclass_payload(result), indent=2, sort_keys=True) + "\n", encoding="utf-8") + harden_private_candidate_tree(output_root) return result @@ -412,6 +433,95 @@ def module_entry_repo(entry: dict[str, Any]) -> str | None: return str(package).split("[", 1)[0] if isinstance(package, str) and package.startswith("govoplan-") else None +def apply_python_artifact_identities( + payload: dict[str, Any], + *, + repo_versions: dict[str, str], + python_artifacts: dict[str, Path | str], +) -> list[CatalogEntryChange]: + """Replace selected release identities with hashes computed from built wheels. + + A version update without a corresponding built artifact deliberately removes + any stale identity for that package. Callers can still prepare a source-only + candidate, but it cannot later establish catalog-anchored installed origin. + """ + + unknown = sorted(set(python_artifacts) - set(repo_versions)) + if unknown: + raise ValueError( + "Built Python artifacts were supplied for unselected repositories: " + + ", ".join(unknown) + ) + package_by_repo: dict[str, str] = {} + core = payload.get("core_release") + if isinstance(core, dict): + package = core.get("python_package") + if isinstance(package, str): + package_by_repo["govoplan-core"] = package + modules = payload.get("modules") + if isinstance(modules, list): + for entry in modules: + if not isinstance(entry, dict): + continue + repo = module_entry_repo(entry) + package = entry.get("python_package") + if repo in repo_versions and isinstance(package, str): + package_by_repo[repo] = package + unmapped_artifacts = sorted(set(python_artifacts) - set(package_by_repo)) + if unmapped_artifacts: + raise ValueError( + "Built artifacts have no catalog Python package mapping: " + + ", ".join(unmapped_artifacts) + ) + + release = payload.get("release") + if not isinstance(release, dict): + raise ValueError("Catalog payload has no release object.") + existing = release.get("artifacts", []) + if not isinstance(existing, list) or any(not isinstance(item, dict) for item in existing): + raise ValueError("Catalog release.artifacts must be a list of objects.") + selected_packages = { + package_by_repo[repo] for repo in repo_versions if repo in package_by_repo + } + retained = [ + item + for item in existing + if str(item.get("package_name") or "") not in selected_packages + ] + changes: list[CatalogEntryChange] = [] + for repo, artifact_path in sorted(python_artifacts.items()): + identity = inspect_python_wheel(artifact_path) + expected_package = package_by_repo[repo] + expected_version = repo_versions[repo].removeprefix("v") + if identity.package_name != expected_package or identity.package_version != expected_version: + raise ValueError( + f"Built artifact for {repo} identifies {identity.package_name} " + f"{identity.package_version}, expected {expected_package} {expected_version}." + ) + retained.append(identity.catalog_payload()) + changes.append( + CatalogEntryChange( + repo=repo, + module_id=None, + field="release.artifact", + before=None, + after=identity.archive_sha256, + ) + ) + retained.sort( + key=lambda item: ( + str(item.get("package_name") or ""), + str(item.get("package_version") or ""), + str(item.get("archive_sha256") or ""), + ) + ) + if retained: + release["artifacts"] = retained + else: + release.pop("artifacts", None) + return changes + + def update_field(target: dict[str, Any], *, repo: str, module_id: str | None, field: str, value: str) -> list[CatalogEntryChange]: before = target.get(field) before_text = before if isinstance(before, str) else None @@ -533,7 +643,13 @@ def resolve_output_dir(*, output_dir: Path | str | None, channel: str, generated if output_dir is not None: return Path(output_dir).expanduser() stamp = generated_at.strftime("%Y%m%d-%H%M%S") - return Path.cwd() / "runtime" / "release-candidates" / f"{channel}-{stamp}" + configured_state = os.getenv("XDG_STATE_HOME", "").strip() + state_root = ( + Path(configured_state).expanduser() + if configured_state + else Path.home() / ".local" / "state" + ) + return state_root / "govoplan" / "release-candidates" / f"{channel}-{stamp}" def json_datetime(value: datetime) -> str: diff --git a/tools/release/release-catalog.py b/tools/release/release-catalog.py index 6e5b729..a9510d4 100644 --- a/tools/release/release-catalog.py +++ b/tools/release/release-catalog.py @@ -23,8 +23,25 @@ def main() -> int: selective.add_argument("--channel", default="stable") selective.add_argument("--repo-version", action="append", default=[], metavar="REPO=VERSION", required=True) selective.add_argument("--catalog-signing-key", action="append", default=[], metavar="KEY_ID=PRIVATE_KEY", required=True) + selective.add_argument( + "--python-artifact", + action="append", + default=[], + metavar="REPO=/PATH/TO/WHEEL", + help=( + "Built immutable wheel whose archive and install-stable payload identity " + "will be computed into the signed catalog; may be repeated." + ), + ) selective.add_argument("--base-catalog", help="Catalog path or URL to use as the source. Defaults to local channel, then public channel.") - selective.add_argument("--output-dir", type=Path, help="Candidate output directory. Defaults to runtime/release-candidates/-.") + selective.add_argument( + "--output-dir", + type=Path, + help=( + "Private candidate output directory. Defaults below " + "$XDG_STATE_HOME/govoplan/release-candidates (or ~/.local/state)." + ), + ) selective.add_argument("--public-base-url", default="https://govoplan.add-ideas.de") selective.add_argument("--repository-base", default="git+ssh://git@git.add-ideas.de/add-ideas") selective.add_argument("--source-remote", default="origin", help="Configured Git remote containing immutable source tags.") @@ -74,6 +91,7 @@ def main() -> int: expires_days=args.expires_days, sequence=args.sequence, check_public=not args.skip_public_check, + python_artifacts=parse_repo_paths(tuple(args.python_artifact)), ) payload = to_jsonable(result) if args.json: @@ -144,6 +162,20 @@ def parse_repo_versions(values: tuple[str, ...]) -> dict[str, str]: return result +def parse_repo_paths(values: tuple[str, ...]) -> dict[str, Path]: + result: dict[str, Path] = {} + for value in values: + if "=" not in value: + raise SystemExit(f"--python-artifact must use REPO=/path/to/wheel: {value}") + repo, path_text = value.split("=", 1) + repo = repo.strip() + path_text = path_text.strip() + if not repo or not path_text or repo in result: + raise SystemExit(f"--python-artifact must uniquely use REPO=/path/to/wheel: {value}") + result[repo] = Path(path_text).expanduser() + return result + + def print_text_summary(payload: dict[str, object]) -> None: print("Selective catalog candidate") for key in (