Bind installed releases to signed artifact receipts

This commit is contained in:
2026-07-22 20:42:05 +02:00
parent 4dc0c8d013
commit 9c0650b2ed
13 changed files with 2179 additions and 52 deletions

View File

@@ -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]:

View File

@@ -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",

View File

@@ -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()