"""Configuration schema and loading helpers for id-configurator.""" from __future__ import annotations import base64 import json from pathlib import Path from uuid import UUID import yaml from pydantic import BaseModel, ConfigDict, Field class NamedContent(BaseModel): """A named piece of content, generally base64-encoded (a certificate, a GPG key, ...).""" name: str content: str def _yes_no(value: bool) -> str: return "yes" if value else "no" class AptSource(BaseModel): """A single APT source, in the DEB822 format (see sources.list(5)).""" model_config = ConfigDict(populate_by_name=True) name: str types: list[str] = Field(default_factory=lambda: ["deb"]) uris: list[str] suites: list[str] components: list[str] = Field(default_factory=list) architectures: list[str] = Field(default_factory=list) languages: list[str] = Field(default_factory=list) targets: list[str] = Field(default_factory=list) enabled: bool = True trusted: bool | None = None signed_by: str | None = Field(default=None, alias="signed-by") pdiffs: bool | None = None by_hash: bool | None = Field(default=None, alias="by-hash") allow_insecure: bool | None = Field(default=None, alias="allow-insecure") allow_weak: bool | None = Field(default=None, alias="allow-weak") allow_downgrade_to_insecure: bool | None = Field( default=None, alias="allow-downgrade-to-insecure" ) check_valid_until: bool | None = Field(default=None, alias="check-valid-until") valid_until_min: str | None = Field(default=None, alias="valid-until-min") valid_until_max: str | None = Field(default=None, alias="valid-until-max") check_date: bool | None = Field(default=None, alias="check-date") date_max_future: str | None = Field(default=None, alias="date-max-future") description: str | None = None def to_deb822(self) -> str: """Render this source as a DEB822 stanza (as used in /etc/apt/sources.list.d/*.sources).""" fields = [ ("Types", " ".join(self.types)), ("URIs", " ".join(self.uris)), ("Suites", " ".join(self.suites)), ] if self.components: fields.append(("Components", " ".join(self.components))) if self.architectures: fields.append(("Architectures", " ".join(self.architectures))) if self.languages: fields.append(("Languages", " ".join(self.languages))) if self.targets: fields.append(("Targets", " ".join(self.targets))) if self.signed_by: fields.append(("Signed-By", self.signed_by)) if self.trusted is not None: fields.append(("Trusted", _yes_no(self.trusted))) if self.pdiffs is not None: fields.append(("PDiffs", _yes_no(self.pdiffs))) if self.by_hash is not None: fields.append(("By-Hash", _yes_no(self.by_hash))) if self.allow_insecure is not None: fields.append(("Allow-Insecure", _yes_no(self.allow_insecure))) if self.allow_weak is not None: fields.append(("Allow-Weak", _yes_no(self.allow_weak))) if self.allow_downgrade_to_insecure is not None: fields.append( ( "Allow-Downgrade-To-Insecure", _yes_no(self.allow_downgrade_to_insecure), ) ) if self.check_valid_until is not None: fields.append(("Check-Valid-Until", _yes_no(self.check_valid_until))) if self.valid_until_min: fields.append(("Valid-Until-Min", self.valid_until_min)) if self.valid_until_max: fields.append(("Valid-Until-Max", self.valid_until_max)) if self.check_date is not None: fields.append(("Check-Date", _yes_no(self.check_date))) if self.date_max_future: fields.append(("Date-Max-Future", self.date_max_future)) if self.description: fields.append(("Description", self.description)) fields.append(("Enabled", _yes_no(self.enabled))) return "\n".join(f"{key}: {value}" for key, value in fields) + "\n" class AptConfig(BaseModel): sources: list[AptSource] = Field(default_factory=list) keys: list[NamedContent] = Field(default_factory=list) class ApkRepository(BaseModel): """An Alpine APK repository (see /etc/apk/repositories), e.g. a URL ending in /main or /community.""" url: str key: str | None = None class ApkConfig(BaseModel): repositories: list[ApkRepository] = Field(default_factory=list) class YumRepository(BaseModel): model_config = ConfigDict(populate_by_name=True) name: str index_url: str = Field(alias="index-url") gpg_url: str | None = Field(default=None, alias="gpg-url") class YumConfig(BaseModel): repositories: list[YumRepository] = Field(default_factory=list) keys: list[NamedContent] = Field(default_factory=list) class PythonRepository(BaseModel): model_config = ConfigDict(populate_by_name=True) name: str index_url: str = Field(alias="index-url") class PythonConfig(BaseModel): repositories: list[PythonRepository] = Field(default_factory=list) class JavaRepository(BaseModel): """A Maven-compatible repository, shared by Maven, Gradle, Ivy, SBT and Leiningen.""" name: str url: str class JavaConfig(BaseModel): repositories: list[JavaRepository] = Field(default_factory=list) class ConfigurationsBlock(BaseModel): certificates: list[NamedContent] = Field(default_factory=list) apt: AptConfig = Field(default_factory=AptConfig) apk: ApkConfig = Field(default_factory=ApkConfig) yum: YumConfig = Field(default_factory=YumConfig) python: PythonConfig = Field(default_factory=PythonConfig) java: JavaConfig = Field(default_factory=JavaConfig) class Configuration(BaseModel): uuid: UUID organisation: str description: str = "" configurations: ConfigurationsBlock = Field(default_factory=ConfigurationsBlock) def parse_configuration(raw: str) -> Configuration: """Parse a configuration from a JSON or YAML string.""" try: data = json.loads(raw) except json.JSONDecodeError: data = yaml.safe_load(raw) return Configuration.model_validate(data) def load_configuration( config_file: Path | None, config_env: str | None ) -> Configuration: """Load a configuration from a file, or from a base64-encoded string (typically an env var).""" if config_file is not None: raw = config_file.read_text(encoding="utf-8") elif config_env is not None: raw = base64.b64decode(config_env).decode("utf-8") else: raise ValueError("either --config or --config-env must be provided") return parse_configuration(raw)