#!/usr/bin/env python3 """Generate and sign a GovOPlaN module package release catalog.""" from __future__ import annotations import argparse import base64 from dataclasses import dataclass from datetime import UTC, datetime, timedelta import json import os from pathlib import Path import sys from typing import Any from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey META_ROOT = Path(__file__).resolve().parents[2] CORE_ROOT = Path(os.environ.get("GOVOPLAN_CORE_ROOT", META_ROOT.parent / "govoplan-core")).resolve() sys.path.insert(0, str(CORE_ROOT / "src")) from govoplan_core.core.modules import ModuleManifest # noqa: E402 from govoplan_core.server.registry import available_module_manifests # noqa: E402 GITEA_BASE = "git+ssh://git@git.add-ideas.de/add-ideas" @dataclass(frozen=True, slots=True) class CatalogModule: module_id: str repo: str python_package: str name: str description: str tags: tuple[str, ...] webui_package: str | None = None provides_interfaces: tuple[dict[str, object], ...] = () requires_interfaces: tuple[dict[str, object], ...] = () CATALOG_MODULES = ( CatalogModule( module_id="tenancy", repo="govoplan-tenancy", python_package="govoplan-tenancy", name="Tenancy", description="Tenant registry, tenant settings, and tenant resolution platform module.", tags=("official", "platform-module"), ), CatalogModule( module_id="organizations", repo="govoplan-organizations", python_package="govoplan-organizations", name="Organizations", description="Organization units, functions, and account-held function assignments.", tags=("official", "platform-module"), ), CatalogModule( module_id="identity", repo="govoplan-identity", python_package="govoplan-identity", name="Identity", description="Canonical identities and links between identities and platform accounts.", tags=("official", "platform-module"), ), CatalogModule( module_id="access", repo="govoplan-access", python_package="govoplan-access", name="Access", description="Authentication, accounts, users, groups, roles, API keys, and access capabilities.", tags=("official", "platform-module"), webui_package="@govoplan/access-webui", ), CatalogModule( module_id="admin", repo="govoplan-admin", python_package="govoplan-admin", name="Admin", description="System settings, governance templates, module management, and admin shell contributions.", tags=("official", "platform-module"), webui_package="@govoplan/admin-webui", ), CatalogModule( module_id="policy", repo="govoplan-policy", python_package="govoplan-policy", name="Policy", description="Policy and governance capability module.", tags=("official", "platform-module"), webui_package="@govoplan/policy-webui", ), CatalogModule( module_id="audit", repo="govoplan-audit", python_package="govoplan-audit", name="Audit", description="Audit-log storage and audit administration routes.", tags=("official", "platform-module"), webui_package="@govoplan/audit-webui", ), CatalogModule( module_id="dashboard", repo="govoplan-dashboard", python_package="govoplan-dashboard", name="Dashboard", description="Configurable user home assembled from module-provided dashboard widgets.", tags=("official", "platform-module"), webui_package="@govoplan/dashboard-webui", ), CatalogModule( module_id="files", repo="govoplan-files", python_package="govoplan-files", name="Files", description="Managed file spaces and campaign attachment integration.", tags=("official", "service-module"), webui_package="@govoplan/files-webui", ), CatalogModule( module_id="mail", repo="govoplan-mail", python_package="govoplan-mail", name="Mail", description="SMTP/IMAP profile management, credential policy, and read-only mailbox access.", tags=("official", "service-module"), webui_package="@govoplan/mail-webui", ), CatalogModule( module_id="campaigns", repo="govoplan-campaign", python_package="govoplan-campaign", name="Campaigns", description="Campaign authoring, validation, queueing, delivery control, and reports.", tags=("official", "business-module"), webui_package="@govoplan/campaign-webui", ), CatalogModule( module_id="calendar", repo="govoplan-calendar", python_package="govoplan-calendar", name="Calendar", description="Calendar collections, events, CalDAV sources, and calendar WebUI routes.", tags=("official", "service-module"), webui_package="@govoplan/calendar-webui", ), CatalogModule( module_id="docs", repo="govoplan-docs", python_package="govoplan-docs", name="Docs", description="Configured-system documentation and evidence-aware help surfaces.", tags=("official", "platform-module"), webui_package="@govoplan/docs-webui", ), CatalogModule( module_id="ops", repo="govoplan-ops", python_package="govoplan-ops", name="Ops", description="Runtime health, deployment profile, worker split, and sizing visibility.", tags=("official", "platform-module"), webui_package="@govoplan/ops-webui", ), ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--version", required=True, help="GovOPlaN release version, without leading v.") parser.add_argument("--channel", default="stable") parser.add_argument("--sequence", type=int, help="Monotonic channel sequence. Defaults to UTC timestamp.") parser.add_argument("--expires-days", type=int, default=90) parser.add_argument("--catalog-output", type=Path, required=True) parser.add_argument("--keyring-output", type=Path) parser.add_argument( "--catalog-signing-key", action="append", default=[], metavar="KEY_ID=PRIVATE_KEY", help="Ed25519 private key used to sign the catalog; may be repeated for rotation.", ) parser.add_argument("--public-base-url", default="https://govoplan.add-ideas.de") parser.add_argument("--repository-base", default=GITEA_BASE) args = parser.parse_args() version = args.version.removeprefix("v") tag = f"v{version}" generated_at = datetime.now(tz=UTC) sequence = args.sequence if args.sequence is not None else int(generated_at.strftime("%Y%m%d%H%M")) expires_at = generated_at + timedelta(days=args.expires_days) signing_keys = [_parse_signing_key(value) for value in args.catalog_signing_key] catalog = _catalog_payload( version=version, tag=tag, channel=args.channel, sequence=sequence, generated_at=generated_at, expires_at=expires_at, repository_base=args.repository_base.rstrip("/"), public_base_url=args.public_base_url.rstrip("/"), ) if signing_keys: catalog["signatures"] = [_signature(catalog, key_id=key_id, private_key=private_key) for key_id, private_key in signing_keys] output = args.catalog_output.expanduser() output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(catalog, indent=2, sort_keys=True) + "\n", encoding="utf-8") if args.keyring_output is not None: keyring_output = args.keyring_output.expanduser() keyring_output.parent.mkdir(parents=True, exist_ok=True) keyring_output.write_text( json.dumps(_keyring(signing_keys=signing_keys, generated_at=generated_at), indent=2, sort_keys=True) + "\n", encoding="utf-8", ) print(f"catalog={output}") if args.keyring_output is not None: print(f"keyring={args.keyring_output.expanduser()}") print(f"channel={args.channel}") print(f"sequence={sequence}") print(f"version={version}") return 0 def _catalog_payload( *, version: str, tag: str, channel: str, sequence: int, generated_at: datetime, expires_at: datetime, repository_base: str, public_base_url: str, ) -> dict[str, Any]: manifests = _discovered_catalog_manifests() modules: list[dict[str, Any]] = [] for module in CATALOG_MODULES: manifest = manifests.get(module.module_id) module_version = manifest.version if manifest is not None else version module_tag = f"v{module_version.removeprefix('v')}" entry: dict[str, Any] = { "module_id": module.module_id, "name": module.name, "description": module.description, "version": module_version, "action": "install", "python_package": module.python_package, "python_ref": f"{module.python_package} @ {repository_base}/{module.repo}.git@{module_tag}", "license_features": [f"module.{module.module_id}"], "tags": list(module.tags), } if module.webui_package: entry["webui_package"] = module.webui_package entry["webui_ref"] = f"{repository_base}/{module.repo}.git#{module_tag}" manifest_metadata = _manifest_catalog_metadata(manifest) entry.update(manifest_metadata) if module.provides_interfaces: entry["provides_interfaces"] = [dict(item) for item in module.provides_interfaces] if module.requires_interfaces: entry["requires_interfaces"] = [dict(item) for item in module.requires_interfaces] modules.append(entry) return { "catalog_version": "1", "channel": channel, "sequence": sequence, "generated_at": _json_datetime(generated_at), "expires_at": _json_datetime(expires_at), "release": { "version": version, "tag": tag, "catalog_url": f"{public_base_url}/catalogs/v1/channels/{channel}.json", "keyring_url": f"{public_base_url}/catalogs/v1/keyring.json", }, "core_release": { "name": "GovOPlaN Core", "version": version, "python_package": "govoplan-core", "python_ref": f"govoplan-core[server] @ {repository_base}/govoplan-core.git@{tag}", "webui_package": "@govoplan/core-webui", "webui_ref": f"{repository_base}/govoplan-core.git#{tag}", }, "modules": modules, } def _discovered_catalog_manifests() -> dict[str, ModuleManifest]: try: return available_module_manifests(ignore_load_errors=True) except Exception: return {} def _manifest_catalog_metadata(manifest: ModuleManifest | None) -> dict[str, object]: if manifest is None: return {} payload: dict[str, object] = {} if manifest.dependencies: payload["dependencies"] = list(manifest.dependencies) if manifest.optional_dependencies: payload["optional_dependencies"] = list(manifest.optional_dependencies) if manifest.migration_spec is not None: payload["migration_safety"] = "requires_review" payload["migration_notes"] = "Module owns database migrations; review release notes and migration output before activation." if manifest.migration_spec.migration_after: payload["migration_after"] = list(manifest.migration_spec.migration_after) if manifest.migration_spec.migration_before: payload["migration_before"] = list(manifest.migration_spec.migration_before) if manifest.migration_spec.migration_tasks: tasks: list[dict[str, object]] = [] for task in manifest.migration_spec.migration_tasks: task_payload: dict[str, object] = { "task_id": task.task_id, "phase": task.phase, "summary": task.summary, "task_version": task.task_version, "safety": task.safety, "idempotent": task.idempotent, } if task.timeout_seconds is not None: task_payload["timeout_seconds"] = task.timeout_seconds tasks.append(task_payload) payload["migration_tasks"] = tasks if manifest.provides_interfaces: payload["provides_interfaces"] = [ {"name": item.name, "version": item.version} for item in manifest.provides_interfaces ] if manifest.requires_interfaces: requirements: list[dict[str, object]] = [] for item in manifest.requires_interfaces: requirement: dict[str, object] = { "name": item.name, "optional": item.optional, } if item.version_min is not None: requirement["version_min"] = item.version_min if item.version_max_exclusive is not None: requirement["version_max_exclusive"] = item.version_max_exclusive requirements.append(requirement) payload["requires_interfaces"] = requirements return payload 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 SystemExit("--catalog-signing-key must use KEY_ID=/path/to/private.pem") path = Path(path_text).expanduser() private_key = serialization.load_pem_private_key(path.read_bytes(), password=None) if not isinstance(private_key, Ed25519PrivateKey): raise SystemExit(f"Catalog signing key must be an Ed25519 private key: {path}") return key_id.strip(), private_key def _signature(payload: dict[str, Any], *, key_id: str, private_key: Ed25519PrivateKey) -> dict[str, str]: signature_payload = dict(payload) signature_payload.pop("signature", None) signature_payload.pop("signatures", None) signature = private_key.sign(_canonical_bytes(signature_payload)) return { "algorithm": "ed25519", "key_id": key_id, "value": base64.b64encode(signature).decode("ascii"), } def _keyring(*, signing_keys: list[tuple[str, Ed25519PrivateKey]], generated_at: datetime) -> dict[str, Any]: return { "keyring_version": "1", "purpose": "govoplan module package catalog signatures", "generated_at": _json_datetime(generated_at), "keys": [ { "key_id": key_id, "status": "active", "public_key": _public_key_base64(private_key), "not_before": generated_at.date().isoformat() + "T00:00:00Z", } for key_id, private_key in signing_keys ], } 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 _json_datetime(value: datetime) -> str: return value.astimezone(UTC).isoformat().replace("+00:00", "Z") if __name__ == "__main__": raise SystemExit(main())