347 lines
14 KiB
Python
347 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate a signed registry-backed GovOPlaN module package catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
from datetime import UTC, datetime, timedelta
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[2]
|
|
CORE_ROOT = Path(os.environ.get("GOVOPLAN_CORE_ROOT", META_ROOT.parent / "govoplan-core")).resolve()
|
|
sys.path.insert(0, str(CORE_ROOT / "src"))
|
|
sys.path.insert(0, str(META_ROOT / "tools" / "release"))
|
|
|
|
from govoplan_release.catalog_entry_synthesis import ( # noqa: E402
|
|
synthesize_repository_catalog_entries,
|
|
validate_initial_entry_closure,
|
|
)
|
|
|
|
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--version", required=True, help="Core release version, without leading v.")
|
|
parser.add_argument("--package-set", type=Path, required=True)
|
|
parser.add_argument("--package-lock", type=Path, required=True)
|
|
parser.add_argument("--workspace", type=Path, default=META_ROOT.parent)
|
|
parser.add_argument("--channel", default="stable")
|
|
parser.add_argument("--sequence", type=int, help="Monotonic channel sequence. Defaults to UTC timestamp.")
|
|
parser.add_argument("--expires-days", type=int, default=90)
|
|
parser.add_argument("--catalog-output", type=Path, required=True)
|
|
parser.add_argument("--keyring-output", type=Path)
|
|
parser.add_argument(
|
|
"--catalog-signing-key",
|
|
action="append",
|
|
default=[],
|
|
metavar="KEY_ID=PRIVATE_KEY",
|
|
help="Ed25519 private key used to sign the catalog; may be repeated for rotation.",
|
|
)
|
|
parser.add_argument("--public-base-url", default="https://govoplan.add-ideas.de")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
version = args.version.removeprefix("v")
|
|
package_set = _read_hashed_json(args.package_set, hash_field="package_set_sha256")
|
|
package_lock = _read_hashed_json(args.package_lock, hash_field="lock_sha256")
|
|
_validate_release_inputs(package_set, package_lock, core_version=version)
|
|
signing_keys = [_parse_signing_key(value) for value in args.catalog_signing_key]
|
|
generated_at = datetime.now(tz=UTC)
|
|
sequence = args.sequence if args.sequence is not None else int(generated_at.strftime("%Y%m%d%H%M"))
|
|
catalog = _catalog_payload(
|
|
package_set=package_set,
|
|
package_lock=package_lock,
|
|
channel=args.channel,
|
|
sequence=sequence,
|
|
generated_at=generated_at,
|
|
expires_at=generated_at + timedelta(days=args.expires_days),
|
|
workspace=args.workspace.expanduser().resolve(),
|
|
public_base_url=args.public_base_url.rstrip("/"),
|
|
)
|
|
if signing_keys:
|
|
catalog["signatures"] = [
|
|
_signature(catalog, key_id=key_id, private_key=private_key)
|
|
for key_id, private_key in signing_keys
|
|
]
|
|
except (KeyError, OSError, ValueError, json.JSONDecodeError) as exc:
|
|
parser.error(str(exc))
|
|
|
|
output = args.catalog_output.expanduser()
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(catalog, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
if args.keyring_output is not None:
|
|
keyring_output = args.keyring_output.expanduser()
|
|
keyring_output.parent.mkdir(parents=True, exist_ok=True)
|
|
keyring_output.write_text(
|
|
json.dumps(_keyring(signing_keys=signing_keys, generated_at=generated_at), indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
print(f"catalog={output}")
|
|
if args.keyring_output is not None:
|
|
print(f"keyring={args.keyring_output.expanduser()}")
|
|
print(f"channel={args.channel}")
|
|
print(f"sequence={sequence}")
|
|
print(f"version={version}")
|
|
print(f"profile={package_set.get('profile', 'base')}")
|
|
return 0
|
|
|
|
|
|
def _catalog_payload(
|
|
*,
|
|
package_set: dict[str, Any],
|
|
package_lock: dict[str, Any],
|
|
channel: str,
|
|
sequence: int,
|
|
generated_at: datetime,
|
|
expires_at: datetime,
|
|
workspace: Path,
|
|
public_base_url: str,
|
|
) -> dict[str, Any]:
|
|
python_lock = _rows_by_name(package_lock, "python")
|
|
webui_lock = _rows_by_repository(package_lock, "webui")
|
|
modules: list[dict[str, object]] = []
|
|
core_release: dict[str, object] | None = None
|
|
selected_units: list[dict[str, str]] = []
|
|
|
|
for package in package_set["python"]:
|
|
name = str(package["name"])
|
|
version = str(package["version"])
|
|
repository = str(package["repository"])
|
|
selected_units.append(
|
|
{
|
|
"repo": repository,
|
|
"version": version,
|
|
"tag": str(package["tag"]),
|
|
"commit": str(package["commit"]),
|
|
}
|
|
)
|
|
python_artifact = python_lock[name]
|
|
webui_artifact = webui_lock.get(repository)
|
|
if name == "govoplan-core":
|
|
core_release = {
|
|
"name": "GovOPlaN Core",
|
|
"version": version,
|
|
"python_package": name,
|
|
"python_ref": _python_ref(name, python_artifact, extras=tuple(package.get("extras") or ())),
|
|
"artifact_integrity": {
|
|
"python": _artifact_integrity(python_artifact, ref=_python_ref(name, python_artifact, extras=tuple(package.get("extras") or ())))
|
|
},
|
|
}
|
|
if webui_artifact is not None:
|
|
webui_ref = _artifact_url(webui_artifact)
|
|
core_release.update(
|
|
{
|
|
"webui_package": webui_artifact["name"],
|
|
"webui_ref": webui_ref,
|
|
}
|
|
)
|
|
core_release["artifact_integrity"]["webui"] = _artifact_integrity(webui_artifact, ref=webui_ref)
|
|
continue
|
|
|
|
entries = synthesize_repository_catalog_entries(
|
|
repo=repository,
|
|
version=version,
|
|
workspace=workspace,
|
|
repository_base="git+https://git.add-ideas.de/GovOPlaN",
|
|
source_ref=str(package["tag"]),
|
|
)
|
|
for entry in entries:
|
|
python_ref = _python_ref(name, python_artifact)
|
|
entry["python_ref"] = python_ref
|
|
entry["source"] = {
|
|
"repository": repository,
|
|
"tag": package["tag"],
|
|
"commit": package["commit"],
|
|
}
|
|
integrity: dict[str, object] = {
|
|
"python": _artifact_integrity(python_artifact, ref=python_ref),
|
|
}
|
|
if entry.get("webui_package"):
|
|
if webui_artifact is None or webui_artifact.get("name") != entry["webui_package"]:
|
|
raise ValueError(f"Package lock has no matching WebUI artifact for {repository}.")
|
|
webui_ref = _artifact_url(webui_artifact)
|
|
entry["webui_ref"] = webui_ref
|
|
integrity["webui"] = _artifact_integrity(webui_artifact, ref=webui_ref)
|
|
else:
|
|
entry.pop("webui_ref", None)
|
|
entry["artifact_integrity"] = integrity
|
|
modules.append(entry)
|
|
|
|
if core_release is None:
|
|
raise ValueError("Package set does not contain govoplan-core.")
|
|
validate_initial_entry_closure(
|
|
catalog_modules=modules,
|
|
initial_module_ids={str(item["module_id"]) for item in modules},
|
|
)
|
|
release_version = str(package_set["release_version"])
|
|
return {
|
|
"catalog_version": "1",
|
|
"channel": channel,
|
|
"sequence": sequence,
|
|
"generated_at": _json_datetime(generated_at),
|
|
"expires_at": _json_datetime(expires_at),
|
|
"release": {
|
|
"version": release_version,
|
|
"tag": f"v{release_version}",
|
|
"profile": package_set.get("profile", "base"),
|
|
"catalog_url": f"{public_base_url}/catalogs/v1/channels/{channel}.json",
|
|
"keyring_url": f"{public_base_url}/catalogs/v1/keyring.json",
|
|
"package_set_sha256": package_set["package_set_sha256"],
|
|
"package_lock_sha256": package_lock["lock_sha256"],
|
|
"selected_units": sorted(selected_units, key=lambda item: item["repo"]),
|
|
},
|
|
"core_release": core_release,
|
|
"modules": sorted(modules, key=lambda item: str(item["module_id"])),
|
|
}
|
|
|
|
|
|
def _read_hashed_json(path: Path, *, hash_field: str) -> dict[str, Any]:
|
|
payload = json.loads(path.expanduser().read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise ValueError(f"{path} must contain a JSON object.")
|
|
expected = payload.get(hash_field)
|
|
unsigned = dict(payload)
|
|
unsigned.pop(hash_field, None)
|
|
if not isinstance(expected, str) or expected != _canonical_sha256(unsigned):
|
|
raise ValueError(f"{path} {hash_field} does not match its contents.")
|
|
return payload
|
|
|
|
|
|
def _validate_release_inputs(package_set: dict[str, Any], package_lock: dict[str, Any], *, core_version: str) -> None:
|
|
if package_set.get("schema_version") != "1" or package_lock.get("schema_version") != "1":
|
|
raise ValueError("Package set and lock must use schema version 1.")
|
|
if package_set.get("release_version") != core_version or package_lock.get("release_version") != core_version:
|
|
raise ValueError("Package set and lock release versions must match --version.")
|
|
if package_lock.get("package_set_sha256") != package_set.get("package_set_sha256"):
|
|
raise ValueError("Package lock does not belong to the selected package set.")
|
|
if package_lock.get("profile", "base") != package_set.get("profile", "base"):
|
|
raise ValueError("Package set and lock profiles do not match.")
|
|
for group in ("python", "webui"):
|
|
selected = {(item.get("name"), item.get("version"), item.get("repository")) for item in package_set.get(group, ()) if isinstance(item, dict)}
|
|
locked = {(item.get("name"), item.get("version"), item.get("repository")) for item in package_lock.get(group, ()) if isinstance(item, dict)}
|
|
if not selected or selected != locked:
|
|
raise ValueError(f"Package lock does not contain the exact {group} package set.")
|
|
for item in package_lock[group]:
|
|
_artifact_url(item)
|
|
if SHA256.fullmatch(str(item.get("sha256") or "")) is None:
|
|
raise ValueError(f"Package lock has an invalid {group} artifact digest.")
|
|
|
|
|
|
def _rows_by_name(payload: dict[str, Any], group: str) -> dict[str, dict[str, object]]:
|
|
return {str(item["name"]): item for item in payload[group]}
|
|
|
|
|
|
def _rows_by_repository(payload: dict[str, Any], group: str) -> dict[str, dict[str, object]]:
|
|
result: dict[str, dict[str, object]] = {}
|
|
for item in payload[group]:
|
|
repository = str(item["repository"])
|
|
if repository in result:
|
|
raise ValueError(f"Package lock contains multiple {group} artifacts for {repository}.")
|
|
result[repository] = item
|
|
return result
|
|
|
|
|
|
def _artifact_url(artifact: dict[str, object]) -> str:
|
|
value = str(artifact.get("url") or "")
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password or parsed.fragment:
|
|
raise ValueError(f"Package artifact has an unsafe download URL: {value!r}.")
|
|
return value
|
|
|
|
|
|
def _python_ref(name: str, artifact: dict[str, object], *, extras: tuple[object, ...] = ()) -> str:
|
|
extra = f"[{','.join(str(item) for item in extras)}]" if extras else ""
|
|
return f"{name}{extra} @ {_artifact_url(artifact)}#sha256={artifact['sha256']}"
|
|
|
|
|
|
def _artifact_integrity(artifact: dict[str, object], *, ref: str) -> dict[str, object]:
|
|
result: dict[str, object] = {
|
|
"ref": ref,
|
|
"url": _artifact_url(artifact),
|
|
"filename": artifact["filename"],
|
|
"sha256": artifact["sha256"],
|
|
"size": artifact["size"],
|
|
"registry_identity": f"{artifact['name']}@{artifact['version']}",
|
|
"git_ref": artifact["tag"],
|
|
"source_commit": artifact["commit"],
|
|
}
|
|
if artifact.get("integrity"):
|
|
result["integrity"] = artifact["integrity"]
|
|
return result
|
|
|
|
|
|
def _parse_signing_key(value: str) -> tuple[str, Ed25519PrivateKey]:
|
|
key_id, separator, path_text = value.partition("=")
|
|
if not separator or not key_id.strip() or not path_text.strip():
|
|
raise ValueError("--catalog-signing-key must use KEY_ID=/path/to/private.pem")
|
|
path = Path(path_text).expanduser()
|
|
private_key = serialization.load_pem_private_key(path.read_bytes(), password=None)
|
|
if not isinstance(private_key, Ed25519PrivateKey):
|
|
raise ValueError(f"Catalog signing key must be an Ed25519 private key: {path}")
|
|
return key_id.strip(), private_key
|
|
|
|
|
|
def _signature(payload: dict[str, Any], *, key_id: str, private_key: Ed25519PrivateKey) -> dict[str, str]:
|
|
signature_payload = dict(payload)
|
|
signature_payload.pop("signature", None)
|
|
signature_payload.pop("signatures", None)
|
|
return {
|
|
"algorithm": "ed25519",
|
|
"key_id": key_id,
|
|
"value": base64.b64encode(private_key.sign(_canonical_bytes(signature_payload))).decode("ascii"),
|
|
}
|
|
|
|
|
|
def _keyring(*, signing_keys: list[tuple[str, Ed25519PrivateKey]], generated_at: datetime) -> dict[str, Any]:
|
|
return {
|
|
"keyring_version": "1",
|
|
"purpose": "govoplan module package catalog signatures",
|
|
"generated_at": _json_datetime(generated_at),
|
|
"keys": [
|
|
{
|
|
"key_id": key_id,
|
|
"status": "active",
|
|
"public_key": base64.b64encode(
|
|
private_key.public_key().public_bytes(
|
|
encoding=serialization.Encoding.Raw,
|
|
format=serialization.PublicFormat.Raw,
|
|
)
|
|
).decode("ascii"),
|
|
"not_before": generated_at.date().isoformat() + "T00:00:00Z",
|
|
}
|
|
for key_id, private_key in signing_keys
|
|
],
|
|
}
|
|
|
|
|
|
def _canonical_bytes(payload: object) -> bytes:
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def _canonical_sha256(payload: object) -> str:
|
|
return hashlib.sha256(_canonical_bytes(payload)).hexdigest()
|
|
|
|
|
|
def _json_datetime(value: datetime) -> str:
|
|
return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|