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