"""Selective release catalog candidate generation.""" from __future__ import annotations import base64 from dataclasses import dataclass from datetime import UTC, datetime, timedelta import json import os from pathlib import Path import re import stat import tempfile from typing import Any from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) 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, ) from .contracts import collect_contracts from .git_state import collect_repository_snapshot, git_text from .model import CatalogEntryChange, SelectiveCatalogCandidate from .module_directory import write_module_directory from .source_provenance import ( SourceTagProvenanceIssue, catalog_source_selection, selected_source_provenance, source_tag_provenance_issues, ) from .version_alignment import ( selected_release_webui_bundle_issues, selected_repository_version_issues, ) from .workspace import ( load_repository_specs, resolve_repo_path, resolve_workspace_root, website_root, ) GITEA_BASE = "git+ssh://git@git.add-ideas.de/add-ideas" MAX_BASE_JSON_BYTES = 16 * 1024 * 1024 _SIGNING_KEY_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$") @dataclass(frozen=True, slots=True) class AuthenticatedCatalogBase: catalog_source: Path | str keyring_source: Path | str catalog: dict[str, Any] keyring: dict[str, Any] def build_selective_catalog_candidate( *, repo_versions: dict[str, str], channel: str = "stable", workspace_root: Path | str | None = None, output_dir: Path | str | None = None, base_catalog: Path | str | None = None, base_keyring: Path | str | None = None, signing_keys: tuple[str, ...] = (), public_base_url: str = DEFAULT_PUBLIC_BASE_URL, repository_base: str = GITEA_BASE, source_remote: str = "origin", expires_days: int = 90, sequence: int | None = None, check_public: bool = True, write_summary: bool = True, python_artifacts: dict[str, Path | str] | None = None, frozen_source_provenance: dict[str, dict[str, 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) if not parsed_keys: raise ValueError("At least one signing key is required.") signer_public_keys = configured_signer_public_keys(parsed_keys) workspace = resolve_workspace_root(workspace_root) enforce_selected_version_alignment(repo_versions=repo_versions, workspace=workspace) if frozen_source_provenance is None: enforce_selected_source_provenance( repo_versions=repo_versions, workspace=workspace, remote=source_remote, ) selected_provenance = selected_source_provenance( repo_versions=repo_versions, workspace=workspace, ) else: selected_provenance = enforce_frozen_selected_source_provenance( repo_versions=repo_versions, expected_provenance=frozen_source_provenance, workspace=workspace, remote=source_remote, ) web_root = website_root(workspace) authenticated_base = load_authenticated_catalog_base( base_catalog=base_catalog, base_keyring=base_keyring, web_root=web_root, channel=channel, public_base_url=public_base_url, signer_public_keys=signer_public_keys, ) source_catalog = authenticated_base.catalog_source payload = authenticated_base.catalog generated_at = datetime.now(tz=UTC) resolved_sequence = sequence or next_sequence(payload, generated_at=generated_at) candidate = json.loads(json.dumps(payload)) candidate["channel"] = channel candidate["sequence"] = resolved_sequence candidate["generated_at"] = json_datetime(generated_at) candidate["expires_at"] = json_datetime(generated_at + timedelta(days=expires_days)) release = candidate.get("release") if not isinstance(release, dict): release = {} release["catalog_url"] = f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json" release["keyring_url"] = f"{public_base_url.rstrip('/')}/catalogs/v1/keyring.json" release["selected_units"] = [ { "repo": repo, "version": version, "tag": f"v{version.removeprefix('v')}", **selected_provenance[repo], } for repo, version in sorted(repo_versions.items()) ] candidate["release"] = release repo_contracts = contracts_by_repo( workspace, selected_repositories=set(repo_versions), ) changes = apply_repo_updates( candidate, repo_versions=repo_versions, repo_contracts=repo_contracts, 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, remote=source_remote, require_selected_heads=frozen_source_provenance is None, ) output_root = resolve_output_dir(output_dir=output_dir, channel=channel, generated_at=generated_at) catalog_path = output_root / "channels" / f"{channel}.json" keyring_path = output_root / "keyring.json" summary_path = output_root / "summary.json" if write_summary else None keyring_payload = candidate_keyring_from_authenticated_base( authenticated_base.keyring, signer_public_keys=signer_public_keys, generated_at=generated_at, ) release["keyring_sha256"] = canonical_hash(keyring_payload) candidate.pop("signature", None) candidate.pop("signatures", None) candidate["signatures"] = [signature(candidate, key_id=key_id, private_key=private_key) for key_id, private_key in parsed_keys] 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, output_root=output_root, channel=channel, public_base_url=public_base_url, ) trusted_keys = trusted_keys_from_keyring(keyring_payload) validation = validate_module_package_catalog( catalog_path, require_trusted=True, approved_channels=(channel,), trusted_keys=trusted_keys, ) validation_warnings = tuple(str(item) for item in validation.get("warnings") or ()) validation_error = validation.get("error") candidate_catalog_hash = canonical_hash(candidate) candidate_keyring_hash = canonical_hash(keyring_payload) public_catalog_url = f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json" public_keyring_url = f"{public_base_url.rstrip('/')}/catalogs/v1/keyring.json" published_catalog_hash: str | None = None published_keyring_hash: str | None = None if check_public: published_catalog = fetch_json(public_catalog_url) if published_catalog["ok"]: published_catalog_hash = canonical_hash(published_catalog["payload"]) published_keyring = fetch_json(public_keyring_url) if published_keyring["ok"]: published_keyring_hash = canonical_hash(published_keyring["payload"]) result = SelectiveCatalogCandidate( generated_at=json_datetime(generated_at), channel=channel, status="ready" if validation.get("valid") else "blocked", candidate_dir=str(output_root), catalog_path=str(catalog_path), keyring_path=str(keyring_path), summary_path=str(summary_path) if summary_path is not None else None, source_catalog=str(source_catalog), public_catalog_url=public_catalog_url, public_keyring_url=public_keyring_url, sequence=resolved_sequence, signature_count=len(candidate["signatures"]) if isinstance(candidate.get("signatures"), list) else 0, key_count=len(keyring_payload.get("keys", ())) if isinstance(keyring_payload.get("keys"), list) else 0, candidate_catalog_hash=candidate_catalog_hash, candidate_keyring_hash=candidate_keyring_hash, published_catalog_hash=published_catalog_hash, published_keyring_hash=published_keyring_hash, candidate_matches_published_catalog=compare_hash(candidate_catalog_hash, published_catalog_hash), candidate_matches_published_keyring=compare_hash(candidate_keyring_hash, published_keyring_hash), validation_valid=validation.get("valid") is True, validation_error=str(validation_error) if validation_error else None, validation_warnings=validation_warnings, changes=tuple(changes), notes=( "Candidate catalog preserves unchanged entries from the source catalog.", f"Generated {len(module_directory_files)} module-directory file(s).", "Publishing is intentionally separate from candidate generation.", ), ) 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 def enforce_selected_version_alignment(*, repo_versions: dict[str, str], workspace: Path) -> None: issues = ( *selected_repository_version_issues( repo_versions=repo_versions, workspace=workspace, ), *selected_release_webui_bundle_issues( repo_versions=repo_versions, workspace=workspace, ), ) failures = [ f"{issue.repo}: {issue.source}={issue.actual!r}, expected {issue.expected!r} ({issue.message})" for issue in issues ] if failures: raise ValueError("Version alignment gate failed: " + "; ".join(failures)) def enforce_selected_source_provenance( *, repo_versions: dict[str, str], workspace: Path, remote: str = "origin", ) -> None: failures = source_tag_provenance_issues( repo_versions=repo_versions, workspace=workspace, remote=remote, require_head_repos=repo_versions, ) if failures: raise ValueError( "Source tag provenance gate failed: " + "; ".join(issue.describe() for issue in failures) ) def enforce_frozen_selected_source_provenance( *, repo_versions: dict[str, str], expected_provenance: dict[str, dict[str, str]], workspace: Path, remote: str = "origin", ) -> dict[str, dict[str, str]]: """Verify detached, clean source checkouts against durable object receipts. Normal interactive candidate generation still requires a named, current source branch. The durable executor instead operates on isolated clones of exact annotated origin tags, so it supplies the commit and tag object that must be present at each detached HEAD. """ if set(expected_provenance) != set(repo_versions): raise ValueError( "Frozen source provenance must cover exactly the selected repositories." ) expected_commits: dict[str, str] = {} expected_tag_objects: dict[str, str] = {} for repo in sorted(repo_versions): identity = expected_provenance.get(repo) commit = identity.get("commit_sha") if isinstance(identity, dict) else None tag_object = ( identity.get("tag_object_sha") if isinstance(identity, dict) else None ) if ( not isinstance(identity, dict) or set(identity) != {"commit_sha", "tag_object_sha"} or not isinstance(commit, str) or re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", commit) is None or not isinstance(tag_object, str) or re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", tag_object) is None ): raise ValueError( f"Frozen source provenance is malformed for {repo}." ) expected_commits[repo] = commit expected_tag_objects[repo] = tag_object failures = list( source_tag_provenance_issues( repo_versions=repo_versions, workspace=workspace, remote=remote, expected_commits=expected_commits, expected_tag_objects=expected_tag_objects, ) ) specs = {spec.name: spec for spec in load_repository_specs(include_website=False)} for repo, version in sorted(repo_versions.items()): spec = specs.get(repo) if spec is None: continue path = resolve_repo_path(spec, workspace) snapshot = collect_repository_snapshot( spec, workspace_root=workspace, target_tag=f"v{version.removeprefix('v')}", online=False, ) if snapshot.dirty_entries: failures.append( SourceTagProvenanceIssue( repo, f"v{version.removeprefix('v')}", "frozen source worktree is not clean", ) ) head = git_text(path, "rev-parse", "--verify", "HEAD") if head != expected_commits[repo]: failures.append( SourceTagProvenanceIssue( repo, f"v{version.removeprefix('v')}", "frozen source HEAD does not match its durable commit receipt", ) ) if failures: raise ValueError( "Frozen source provenance gate failed: " + "; ".join(issue.describe() for issue in failures) ) observed = selected_source_provenance( repo_versions=repo_versions, workspace=workspace, ) if observed != expected_provenance: raise ValueError( "Frozen source object identities changed during candidate synthesis." ) return observed def enforce_complete_catalog_source_provenance( *, payload: object, workspace: Path, remote: str = "origin", require_selected_heads: bool = True, ) -> None: selection = catalog_source_selection(payload) failures = [*selection.issues] failures.extend( source_tag_provenance_issues( repo_versions=selection.all_versions, workspace=workspace, remote=remote, require_head_repos=( selection.selected_versions if require_selected_heads else () ), expected_commits=selection.selected_commits, expected_tag_objects=selection.selected_tag_objects, ) ) if failures: raise ValueError( "Complete catalog source provenance gate failed: " + "; ".join(issue.describe() for issue in failures) ) def load_authenticated_catalog_base( *, base_catalog: Path | str | None, base_keyring: Path | str | None, web_root: Path, channel: str, public_base_url: str, signer_public_keys: dict[str, str], ) -> AuthenticatedCatalogBase: """Read one base pair once and authenticate it before any synthesis.""" catalog_source, keyring_source = resolve_base_sources( base_catalog=base_catalog, base_keyring=base_keyring, web_root=web_root, channel=channel, public_base_url=public_base_url, ) catalog = read_bounded_json_source(catalog_source, label="base catalog") keyring = read_bounded_json_source(keyring_source, label="base keyring") if not isinstance(catalog, dict) or not isinstance(keyring, dict): raise ValueError("Selective catalog base and keyring must be JSON objects.") base_trusted_keys = authenticate_base_keyring(keyring) release = catalog.get("release") pinned_keyring = ( release.get("keyring_sha256") if isinstance(release, dict) else None ) if pinned_keyring != canonical_hash(keyring): raise ValueError( "Signed base catalog does not pin the exact supplied base keyring." ) authenticate_base_catalog_signatures( catalog, base_trusted_keys=base_trusted_keys, configured_signers=signer_public_keys, ) validation = validate_catalog_object( catalog, approved_channel=channel, signer_public_keys={ key_id: public_key for key_id, public_key in signer_public_keys.items() if base_trusted_keys.get(key_id) == public_key }, ) if validation.get("valid") is not True: raise ValueError( "Authenticated base catalog failed catalog validation: " + str(validation.get("error") or "unknown validation error") ) return AuthenticatedCatalogBase( catalog_source=catalog_source, keyring_source=keyring_source, catalog=catalog, keyring=keyring, ) def resolve_base_sources( *, base_catalog: Path | str | None, base_keyring: Path | str | None, web_root: Path, channel: str, public_base_url: str, ) -> tuple[Path | str, Path | str]: if (base_catalog is None) != (base_keyring is None): raise ValueError("--base-catalog and --base-keyring must be supplied together.") if base_catalog is not None and base_keyring is not None: return _json_source(base_catalog), _json_source(base_keyring) local_root = web_root / "public" / "catalogs" / "v1" local_catalog = local_root / "channels" / f"{channel}.json" local_keyring = local_root / "keyring.json" if local_catalog.exists() and local_keyring.exists(): return local_catalog, local_keyring public_root = public_base_url.rstrip("/") return ( f"{public_root}/catalogs/v1/channels/{channel}.json", f"{public_root}/catalogs/v1/keyring.json", ) def _json_source(value: Path | str) -> Path | str: text = str(value) return text if text.startswith(("http://", "https://")) else Path(text).expanduser() def read_bounded_json_source(source: Path | str, *, label: str) -> object: if isinstance(source, str) and source.startswith(("http://", "https://")): payload = fetch_json(source) if not payload["ok"]: raise ValueError(f"Could not fetch {label}: {payload['error']}") return payload["payload"] encoded = _read_regular_file_without_symlinks(Path(source), label=label) try: return json.loads(encoded.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError(f"{label.capitalize()} is not valid UTF-8 JSON.") from exc def _read_regular_file_without_symlinks( path: Path, *, label: str, require_private_owner: bool = False ) -> bytes: absolute = path.absolute() parts = absolute.parts[1:] if not parts: raise ValueError(f"{label.capitalize()} must name a regular file.") directory_flags = ( os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) ) descriptor = os.open(absolute.anchor or "/", directory_flags) try: for part in parts[:-1]: child = os.open(part, directory_flags, dir_fd=descriptor) os.close(descriptor) descriptor = child file_descriptor = os.open( parts[-1], os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=descriptor, ) except OSError as exc: os.close(descriptor) raise ValueError(f"{label.capitalize()} cannot be opened safely.") from exc os.close(descriptor) try: initial = os.fstat(file_descriptor) if not stat.S_ISREG(initial.st_mode) or initial.st_size > MAX_BASE_JSON_BYTES: raise ValueError( f"{label.capitalize()} must be a bounded regular file." ) if require_private_owner and ( initial.st_uid != os.geteuid() or stat.S_IMODE(initial.st_mode) & 0o077 ): raise ValueError( f"{label.capitalize()} must be owned by the current operator " "and inaccessible to other users." ) chunks: list[bytes] = [] remaining = initial.st_size while remaining: chunk = os.read(file_descriptor, min(1024 * 1024, remaining)) if not chunk: break chunks.append(chunk) remaining -= len(chunk) final = os.fstat(file_descriptor) if remaining or _stat_fingerprint(initial) != _stat_fingerprint(final): raise ValueError(f"{label.capitalize()} changed while it was read.") return b"".join(chunks) finally: os.close(file_descriptor) def _stat_fingerprint(value: os.stat_result) -> tuple[int, int, int, int, int]: return ( value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns, ) def next_sequence(payload: dict[str, Any], *, generated_at: datetime) -> int: timestamp_sequence = int(generated_at.strftime("%Y%m%d%H%M")) try: current = int(payload.get("sequence") or 0) except (TypeError, ValueError): current = 0 return max(current + 1, timestamp_sequence) def contracts_by_repo( workspace: Path, *, selected_repositories: set[str] | None = None ) -> dict[str, Any]: specs = tuple( spec for spec in load_repository_specs(include_website=False) if selected_repositories is None or spec.name in selected_repositories ) snapshots = tuple( collect_repository_snapshot(spec, workspace_root=workspace, target_tag=None, online=False) for spec in specs ) return {contract.repo: contract for contract in collect_contracts(snapshots)} def apply_repo_updates( payload: dict[str, Any], *, repo_versions: dict[str, str], repo_contracts: dict[str, Any], repository_base: str, workspace: Path, ) -> list[CatalogEntryChange]: changes: list[CatalogEntryChange] = [] modules = payload.get("modules") if not isinstance(modules, list): raise ValueError("Catalog payload has no modules list.") handled: set[str] = set() if "govoplan-core" in repo_versions: version = repo_versions["govoplan-core"].removeprefix("v") tag = f"v{version}" core_release = payload.get("core_release") if not isinstance(core_release, dict): raise ValueError("Catalog payload has no core_release object.") changes.extend(update_field(core_release, repo="govoplan-core", module_id=None, field="version", value=version)) changes.extend(update_field(core_release, repo="govoplan-core", module_id=None, field="python_ref", value=f"govoplan-core[server] @ {repository_base}/govoplan-core.git@{tag}")) changes.extend(update_field(core_release, repo="govoplan-core", module_id=None, field="webui_ref", value=f"{repository_base}/govoplan-core.git#{tag}")) release = payload.get("release") if isinstance(release, dict) and isinstance(release.get("version"), str): changes.extend(update_field(release, repo="govoplan-core", module_id=None, field="version", value=version)) changes.extend(update_field(release, repo="govoplan-core", module_id=None, field="tag", value=tag)) handled.add("govoplan-core") for entry in modules: if not isinstance(entry, dict): continue repo = module_entry_repo(entry) if repo not in repo_versions: continue version = repo_versions[repo].removeprefix("v") tag = f"v{version}" module_id = str(entry.get("module_id") or "") package = str(entry.get("python_package") or repo) changes.extend(update_field(entry, repo=repo, module_id=module_id, field="version", value=version)) changes.extend(update_field(entry, repo=repo, module_id=module_id, field="python_ref", value=f"{package} @ {repository_base}/{repo}.git@{tag}")) if entry.get("webui_package"): changes.extend(update_field(entry, repo=repo, module_id=module_id, field="webui_ref", value=f"{repository_base}/{repo}.git#{tag}")) contract = repo_contracts.get(repo) if contract is not None: provider_payload = [{"name": item.name, "version": item.version} for item in contract.provides_interfaces] requirement_payload = [ { "name": item.name, **({"version_min": item.version_min} if item.version_min else {}), **({"version_max_exclusive": item.version_max_exclusive} if item.version_max_exclusive else {}), "optional": item.optional, } for item in contract.requires_interfaces ] if provider_payload: changes.extend(update_json_field(entry, repo=repo, module_id=module_id, field="provides_interfaces", value=provider_payload)) if requirement_payload: changes.extend(update_json_field(entry, repo=repo, module_id=module_id, field="requires_interfaces", value=requirement_payload)) handled.add(repo) missing = sorted(set(repo_versions) - handled) initial_module_ids: set[str] = set() existing_module_ids = { str(entry.get("module_id") or "") for entry in modules if isinstance(entry, dict) } for repo in missing: entries = synthesize_repository_catalog_entries( repo=repo, version=repo_versions[repo], workspace=workspace, repository_base=repository_base, ) for entry in entries: module_id = str(entry["module_id"]) if module_id in existing_module_ids: raise ValueError( f"Cannot synthesize {repo}: module id {module_id!r} already belongs to another catalog entry." ) modules.append(entry) existing_module_ids.add(module_id) initial_module_ids.add(module_id) changes.append( CatalogEntryChange( repo=repo, module_id=module_id, field="catalog_entry", before=None, after=json.dumps(entry, sort_keys=True), ) ) handled.add(repo) if initial_module_ids: validate_initial_entry_closure( catalog_modules=modules, initial_module_ids=initial_module_ids, ) return changes def module_entry_repo(entry: dict[str, Any]) -> str | None: for field in ("python_ref", "webui_ref"): value = entry.get(field) if isinstance(value, str): match = re.search(r"/([^/@#]+)[.]git(?:[@#]|$)", value) if match: return match.group(1) package = entry.get("python_package") 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 if before_text == value: return [] target[field] = value return [CatalogEntryChange(repo=repo, module_id=module_id, field=field, before=before_text, after=value)] def update_json_field(target: dict[str, Any], *, repo: str, module_id: str | None, field: str, value: list[dict[str, object]]) -> list[CatalogEntryChange]: before = target.get(field) if before == value: return [] target[field] = value return [ CatalogEntryChange( repo=repo, module_id=module_id, field=field, before=json.dumps(before, sort_keys=True) if before is not None else None, after=json.dumps(value, sort_keys=True), ) ] 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") key_id = key_id.strip() if _SIGNING_KEY_ID.fullmatch(key_id) is None: raise ValueError("Catalog signing key ID is not a bounded opaque ID.") path = Path(path_text).expanduser() private_key = serialization.load_pem_private_key( _read_regular_file_without_symlinks( path, label="catalog signing key", require_private_owner=True, ), password=None, ) if not isinstance(private_key, Ed25519PrivateKey): raise ValueError(f"Catalog signing key must be an Ed25519 private key: {path}") return key_id, 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 configured_signer_public_keys( signing_keys: tuple[tuple[str, Ed25519PrivateKey], ...], ) -> dict[str, str]: result: dict[str, str] = {} seen_material: set[str] = set() for key_id, private_key in signing_keys: public_key = public_key_base64(private_key) if key_id in result or public_key in seen_material: raise ValueError("Catalog signing keys contain duplicate identity or material.") result[key_id] = public_key seen_material.add(public_key) return result def authenticate_base_keyring(payload: dict[str, Any]) -> dict[str, str]: keys = payload.get("keys") if not isinstance(keys, list) or not keys or len(keys) > 64: raise ValueError("Base keyring has no bounded signer set.") observed_ids: set[str] = set() seen_material: set[bytes] = set() trusted: dict[str, str] = {} for item in keys: if not isinstance(item, dict): raise ValueError("Base keyring contains a malformed key.") key_id = item.get("key_id") public_text = item.get("public_key") status = item.get("status", "active") if ( not isinstance(key_id, str) or _SIGNING_KEY_ID.fullmatch(key_id) is None or not isinstance(public_text, str) or status not in {"active", "next", "retired", "revoked", "disabled"} or key_id in observed_ids ): raise ValueError("Base keyring contains a malformed or duplicate key.") try: public_bytes = base64.b64decode(public_text, validate=True) Ed25519PublicKey.from_public_bytes(public_bytes) except ValueError as exc: raise ValueError("Base keyring contains invalid Ed25519 material.") from exc if public_bytes in seen_material: raise ValueError("Base keyring reuses signer public material.") seen_material.add(public_bytes) observed_ids.add(key_id) if status in {"active", "next"}: trusted[key_id] = base64.b64encode(public_bytes).decode("ascii") if not trusted: raise ValueError("Base keyring has no active trusted catalog signer.") return trusted def authenticate_base_catalog_signatures( payload: dict[str, Any], *, base_trusted_keys: dict[str, str], configured_signers: dict[str, str], ) -> None: if payload.get("signature") is not None: raise ValueError("Base catalog must use one unambiguous signatures envelope.") signatures = payload.get("signatures") if not isinstance(signatures, list) or not signatures or len(signatures) > 16: raise ValueError("Base catalog has no bounded signatures envelope.") signed_payload = dict(payload) signed_payload.pop("signatures", None) encoded = canonical_bytes(signed_payload) seen: set[str] = set() for item in signatures: if not isinstance(item, dict) or item.get("algorithm") != "ed25519": raise ValueError("Base catalog contains a malformed signature.") key_id = item.get("key_id") value = item.get("value") if ( not isinstance(key_id, str) or key_id in seen or key_id not in base_trusted_keys or not isinstance(value, str) ): raise ValueError("Base catalog contains an untrusted or duplicate signature.") try: public_key = Ed25519PublicKey.from_public_bytes( base64.b64decode(base_trusted_keys[key_id], validate=True) ) public_key.verify(base64.b64decode(value, validate=True), encoded) except (ValueError, InvalidSignature) as exc: raise ValueError("Base catalog signature verification failed.") from exc seen.add(key_id) if not any( key_id in seen and configured_signers.get(key_id) == base_trusted_keys.get(key_id) for key_id in configured_signers ): raise ValueError( "Base catalog needs a valid signature from a configured signer " "already trusted by its pinned keyring." ) def candidate_keyring_from_authenticated_base( payload: dict[str, Any], *, signer_public_keys: dict[str, str], generated_at: datetime, ) -> dict[str, Any]: """Carry authenticated keys forward and add only configured signer material.""" candidate = json.loads(json.dumps(payload)) keys = candidate.get("keys") if not isinstance(keys, list): raise ValueError("Authenticated base keyring lost its keys list.") existing: dict[str, str] = {} existing_material: set[str] = set() for item in keys: if not isinstance(item, dict): raise ValueError("Authenticated base keyring contains a malformed key.") key_id = str(item.get("key_id") or "") public_key = str(item.get("public_key") or "") existing[key_id] = public_key existing_material.add(public_key) changed = False for key_id, public_key in sorted(signer_public_keys.items()): if key_id in existing: if existing[key_id] != public_key: raise ValueError( "Configured signer key ID conflicts with authenticated base material." ) matching = next( item for item in keys if isinstance(item, dict) and item.get("key_id") == key_id ) if matching.get("status", "active") not in {"active", "next"}: raise ValueError("Configured signer is disabled in the base keyring.") continue if public_key in existing_material: raise ValueError( "Configured signer material already has another authenticated key ID." ) keys.append( { "key_id": key_id, "status": "active", "public_key": public_key, "not_before": generated_at.astimezone(UTC) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z"), } ) existing_material.add(public_key) changed = True if changed: candidate["generated_at"] = json_datetime(generated_at) return candidate def validate_catalog_object( payload: dict[str, Any], *, approved_channel: str, signer_public_keys: dict[str, str], ) -> dict[str, object]: descriptor, temporary_name = tempfile.mkstemp( prefix="govoplan-base-catalog-", suffix=".json" ) temporary = Path(temporary_name) try: os.fchmod(descriptor, 0o600) encoded = (json.dumps(payload, sort_keys=True) + "\n").encode("utf-8") if len(encoded) > MAX_BASE_JSON_BYTES: raise ValueError("Base catalog exceeds its size limit.") with os.fdopen(descriptor, "wb", closefd=True) as handle: handle.write(encoded) handle.flush() os.fsync(handle.fileno()) return validate_module_package_catalog( temporary, require_trusted=True, approved_channels=(approved_channel,), trusted_keys=signer_public_keys, ) finally: try: os.close(descriptor) except OSError: pass temporary.unlink(missing_ok=True) def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]: keys = payload.get("keys") if not isinstance(keys, list): return {} result: dict[str, str] = {} for item in keys: if not isinstance(item, dict): continue status = str(item.get("status") or "active").lower() if status in {"revoked", "disabled", "retired"}: continue key_id = str(item.get("key_id") or "") public_key = str(item.get("public_key") or item.get("public_key_base64") or "") if key_id and public_key: result[key_id] = public_key return result def public_key_base64(private_key: Ed25519PrivateKey) -> str: public_bytes = private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) return base64.b64encode(public_bytes).decode("ascii") def canonical_bytes(payload: object) -> bytes: return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def compare_hash(candidate_hash: str | None, published_hash: str | None) -> bool | None: if candidate_hash is None or published_hash is None: return None return candidate_hash == published_hash def resolve_output_dir(*, output_dir: Path | str | None, channel: str, generated_at: datetime) -> Path: if output_dir is not None: return Path(output_dir).expanduser() stamp = generated_at.strftime("%Y%m%d-%H%M%S") 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: return value.astimezone(UTC).isoformat().replace("+00:00", "Z") def dataclass_payload(value: object) -> dict[str, object]: from dataclasses import asdict return asdict(value)