987 lines
38 KiB
Python
987 lines
38 KiB
Python
"""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
|
|
from .model import CatalogEntryChange, SelectiveCatalogCandidate
|
|
from .module_directory import write_module_directory
|
|
from .source_provenance import (
|
|
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_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,
|
|
) -> 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)
|
|
enforce_selected_source_provenance(
|
|
repo_versions=repo_versions,
|
|
workspace=workspace,
|
|
remote=source_remote,
|
|
)
|
|
selected_provenance = selected_source_provenance(
|
|
repo_versions=repo_versions,
|
|
workspace=workspace,
|
|
)
|
|
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)
|
|
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,
|
|
)
|
|
|
|
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_complete_catalog_source_provenance(
|
|
*,
|
|
payload: object,
|
|
workspace: Path,
|
|
remote: str = "origin",
|
|
) -> 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,
|
|
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) -> 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."
|
|
)
|
|
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) -> dict[str, Any]:
|
|
specs = load_repository_specs(include_website=False)
|
|
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"),
|
|
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)
|