From fdee766993319cc4765b2b0506d43cebc781ead8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 20:59:00 +0200 Subject: [PATCH] Authenticate selective catalog base pairs --- tests/test_release_base_catalog_trust.py | 184 ++++++++ tests/test_release_source_provenance.py | 50 ++- .../govoplan_release/selective_catalog.py | 414 ++++++++++++++++-- tools/release/release-catalog.py | 16 +- 4 files changed, 602 insertions(+), 62 deletions(-) create mode 100644 tests/test_release_base_catalog_trust.py diff --git a/tests/test_release_base_catalog_trust.py b/tests/test_release_base_catalog_trust.py new file mode 100644 index 0000000..7d02ef7 --- /dev/null +++ b/tests/test_release_base_catalog_trust.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +import sys +import tempfile +import unittest + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +META_ROOT = Path(__file__).resolve().parents[1] +RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" +if str(RELEASE_TOOLS_ROOT) not in sys.path: + sys.path.insert(0, str(RELEASE_TOOLS_ROOT)) + +from govoplan_release.catalog import canonical_hash # noqa: E402 +from govoplan_release.selective_catalog import ( # noqa: E402 + candidate_keyring_from_authenticated_base, + load_authenticated_catalog_base, + public_key_base64, + signature, +) + + +class ReleaseBaseCatalogTrustTests(unittest.TestCase): + def test_exact_signed_base_pair_is_loaded_once_and_carried_in_memory(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + private_key = Ed25519PrivateKey.generate() + catalog, keyring = self._write_pair(root, private_key=private_key) + + authenticated = load_authenticated_catalog_base( + base_catalog=catalog, + base_keyring=keyring, + web_root=root, + channel="stable", + public_base_url="https://invalid.example", + signer_public_keys={"release-key": public_key_base64(private_key)}, + ) + + keyring.unlink() + + self.assertEqual("release-key", authenticated.keyring["keys"][0]["key_id"]) + self.assertEqual( + canonical_hash(authenticated.keyring), + authenticated.catalog["release"]["keyring_sha256"], + ) + + def test_deliberate_old_to_new_signer_rotation_is_carried_forward(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + private_key = Ed25519PrivateKey.generate() + new_key = Ed25519PrivateKey.generate() + catalog, keyring = self._write_pair(root, private_key=private_key) + configured = { + "release-key": public_key_base64(private_key), + "release-key-next": public_key_base64(new_key), + } + + authenticated = load_authenticated_catalog_base( + base_catalog=catalog, + base_keyring=keyring, + web_root=root, + channel="stable", + public_base_url="https://invalid.example", + signer_public_keys=configured, + ) + candidate = candidate_keyring_from_authenticated_base( + authenticated.keyring, + signer_public_keys=configured, + generated_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + ) + + self.assertEqual( + {"release-key", "release-key-next"}, + {item["key_id"] for item in candidate["keys"]}, + ) + + def test_injected_extra_key_without_catalog_authorization_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + private_key = Ed25519PrivateKey.generate() + extra_key = Ed25519PrivateKey.generate() + catalog, keyring = self._write_pair(root, private_key=private_key) + payload = json.loads(keyring.read_text(encoding="utf-8")) + payload["keys"].append( + { + "key_id": "injected-key", + "status": "active", + "public_key": public_key_base64(extra_key), + } + ) + keyring.write_text(json.dumps(payload), encoding="utf-8") + + with self.assertRaisesRegex(ValueError, "does not pin"): + load_authenticated_catalog_base( + base_catalog=catalog, + base_keyring=keyring, + web_root=root, + channel="stable", + public_base_url="https://invalid.example", + signer_public_keys={ + "release-key": public_key_base64(private_key) + }, + ) + + def test_unpinned_or_symlinked_base_keyring_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + private_key = Ed25519PrivateKey.generate() + catalog, keyring = self._write_pair(root, private_key=private_key) + payload = json.loads(keyring.read_text(encoding="utf-8")) + payload["generated_at"] = "changed" + keyring.write_text(json.dumps(payload), encoding="utf-8") + + with self.assertRaisesRegex(ValueError, "does not pin"): + load_authenticated_catalog_base( + base_catalog=catalog, + base_keyring=keyring, + web_root=root, + channel="stable", + public_base_url="https://invalid.example", + signer_public_keys={ + "release-key": public_key_base64(private_key) + }, + ) + + keyring.unlink() + outside = root / "outside.json" + outside.write_text("{}", encoding="utf-8") + keyring.symlink_to(outside) + with self.assertRaisesRegex(ValueError, "opened safely"): + load_authenticated_catalog_base( + base_catalog=catalog, + base_keyring=keyring, + web_root=root, + channel="stable", + public_base_url="https://invalid.example", + signer_public_keys={ + "release-key": public_key_base64(private_key) + }, + ) + + @staticmethod + def _write_pair( + root: Path, *, private_key: Ed25519PrivateKey + ) -> tuple[Path, Path]: + keys = [ + { + "key_id": "release-key", + "status": "active", + "public_key": public_key_base64(private_key), + } + ] + keyring_payload = { + "keyring_version": "1", + "purpose": "govoplan module package catalog signatures", + "keys": keys, + } + catalog_payload = { + "channel": "stable", + "sequence": 1, + "core_release": {"version": "1.0.0"}, + "modules": [], + "release": {"keyring_sha256": canonical_hash(keyring_payload)}, + } + catalog_payload["signatures"] = [ + signature( + catalog_payload, + key_id="release-key", + private_key=private_key, + ) + ] + catalog = root / "stable.json" + keyring = root / "keyring.json" + catalog.write_text(json.dumps(catalog_payload), encoding="utf-8") + keyring.write_text(json.dumps(keyring_payload), encoding="utf-8") + return catalog, keyring + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_release_source_provenance.py b/tests/test_release_source_provenance.py index bbc949b..bca1370 100644 --- a/tests/test_release_source_provenance.py +++ b/tests/test_release_source_provenance.py @@ -19,9 +19,12 @@ if str(RELEASE_ROOT) not in sys.path: sys.path.insert(0, str(RELEASE_ROOT)) from govoplan_release.publisher import publish_catalog_candidate # noqa: E402 +from govoplan_release.catalog import canonical_hash # noqa: E402 from govoplan_release.selective_catalog import ( # noqa: E402 build_selective_catalog_candidate, enforce_selected_source_provenance, + public_key_base64, + signature, ) from govoplan_release.source_provenance import ( # noqa: E402 catalog_source_selection, @@ -117,23 +120,37 @@ class ReleaseSourceProvenanceTests(unittest.TestCase): core, _ = make_repo(self.workspace, self.root, "govoplan-core", "1.2.3") git(core, "tag", "-a", "v1.2.3", "-m", "Core 1.2.3") git(core, "push", "origin", "refs/tags/v1.2.3") - base_catalog = self.root / "base.json" - base_catalog.write_text( - json.dumps( - { - "channel": "stable", - "sequence": 1, - "core_release": { - "version": "1.2.2", - "python_ref": "govoplan-core @ git+ssh://git@example.test/acme/govoplan-core.git@v1.2.2", - }, - "modules": [], - "release": {"version": "1.2.2", "tag": "v1.2.2"}, - } - ), - encoding="utf-8", - ) private_key = Ed25519PrivateKey.generate() + keyring = { + "keys": [ + { + "key_id": "test-key", + "status": "active", + "public_key": public_key_base64(private_key), + } + ] + } + base_keyring = self.root / "base-keyring.json" + base_keyring.write_text(json.dumps(keyring), encoding="utf-8") + base_catalog = self.root / "base.json" + base_payload = { + "channel": "stable", + "sequence": 1, + "core_release": { + "version": "1.2.2", + "python_ref": "govoplan-core @ git+ssh://git@example.test/acme/govoplan-core.git@v1.2.2", + }, + "modules": [], + "release": { + "version": "1.2.2", + "tag": "v1.2.2", + "keyring_sha256": canonical_hash(keyring), + }, + } + base_payload["signatures"] = [ + signature(base_payload, key_id="test-key", private_key=private_key) + ] + base_catalog.write_text(json.dumps(base_payload), encoding="utf-8") key_path = self.root / "release.pem" key_path.write_bytes( private_key.private_bytes( @@ -153,6 +170,7 @@ class ReleaseSourceProvenanceTests(unittest.TestCase): workspace_root=self.workspace, output_dir=output, base_catalog=base_catalog, + base_keyring=base_keyring, signing_keys=(f"test-key={key_path}",), check_public=False, ) diff --git a/tools/release/govoplan_release/selective_catalog.py b/tools/release/govoplan_release/selective_catalog.py index 1e26aca..3dfdd2e 100644 --- a/tools/release/govoplan_release/selective_catalog.py +++ b/tools/release/govoplan_release/selective_catalog.py @@ -3,15 +3,22 @@ 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 +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) from govoplan_core.core.module_package_catalog import validate_module_package_catalog @@ -41,6 +48,16 @@ from .version_alignment import ( 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( @@ -50,6 +67,7 @@ def build_selective_catalog_candidate( 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, @@ -66,6 +84,7 @@ def build_selective_catalog_candidate( 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) @@ -79,10 +98,16 @@ def build_selective_catalog_candidate( workspace=workspace, ) web_root = website_root(workspace) - source_catalog = resolve_base_catalog(base_catalog=base_catalog, web_root=web_root, channel=channel) - payload = read_catalog(source_catalog) - if not isinstance(payload, dict): - raise ValueError("Selective catalog updates require object-style catalogs.") + 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) @@ -132,9 +157,9 @@ def build_selective_catalog_candidate( 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 = keyring_payload_for_candidate( - existing_keyring=web_root / "public" / "catalogs" / "v1" / "keyring.json", - signing_keys=parsed_keys, + 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) @@ -283,23 +308,161 @@ def enforce_complete_catalog_source_provenance( ) -def resolve_base_catalog(*, base_catalog: Path | str | None, web_root: Path, channel: str) -> Path | str: - if base_catalog is not None: - value = str(base_catalog) - return value if value.startswith(("http://", "https://")) else Path(value).expanduser() - local = web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json" - if local.exists(): - return local - return f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/{channel}.json" +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 read_catalog(source: Path | str) -> object: +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 catalog {source}: {payload['error']}") + raise ValueError(f"Could not fetch {label}: {payload['error']}") return payload["payload"] - return json.loads(Path(source).read_text(encoding="utf-8")) + 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: @@ -551,11 +714,17 @@ 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(path.read_bytes(), password=None) + 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.strip(), private_key + return key_id, private_key def signature(payload: dict[str, Any], *, key_id: str, private_key: Ed25519PrivateKey) -> dict[str, str]: @@ -569,38 +738,193 @@ def signature(payload: dict[str, Any], *, key_id: str, private_key: Ed25519Priva } -def keyring_payload_for_candidate( - *, - existing_keyring: Path, +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]: - expected = {key_id: public_key_base64(private_key) for key_id, private_key in signing_keys} - if existing_keyring.exists(): - payload = json.loads(existing_keyring.read_text(encoding="utf-8")) - keys = payload.get("keys") if isinstance(payload, dict) else None - if isinstance(keys, list): - existing = { - str(item.get("key_id")): str(item.get("public_key") or item.get("public_key_base64")) + """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") - } - if all(existing.get(key_id) == public_key for key_id, public_key in expected.items()): - return payload - return { - "keyring_version": "1", - "purpose": "govoplan module package catalog signatures", - "generated_at": json_datetime(generated_at), - "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_base64(private_key), - "not_before": generated_at.date().isoformat() + "T00:00:00Z", + "public_key": public_key, + "not_before": generated_at.astimezone(UTC) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z"), } - for key_id, private_key in signing_keys - ], - } + ) + 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]: diff --git a/tools/release/release-catalog.py b/tools/release/release-catalog.py index a9510d4..2abe8e6 100644 --- a/tools/release/release-catalog.py +++ b/tools/release/release-catalog.py @@ -33,7 +33,20 @@ def main() -> int: "will be computed into the signed catalog; may be repeated." ), ) - selective.add_argument("--base-catalog", help="Catalog path or URL to use as the source. Defaults to local channel, then public channel.") + selective.add_argument( + "--base-catalog", + help=( + "Authenticated catalog path or URL to use as the source; must be " + "paired with --base-keyring. Defaults to the local pair, then public pair." + ), + ) + selective.add_argument( + "--base-keyring", + help=( + "Exact keyring path or URL pinned by --base-catalog and matching " + "the configured catalog signer set." + ), + ) selective.add_argument( "--output-dir", type=Path, @@ -84,6 +97,7 @@ def main() -> int: workspace_root=args.workspace_root, output_dir=args.output_dir, base_catalog=args.base_catalog, + base_keyring=args.base_keyring, signing_keys=tuple(args.catalog_signing_key), public_base_url=args.public_base_url, repository_base=args.repository_base,