Files
govoplan/tests/test_release_artifact_identity.py

278 lines
11 KiB
Python

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