"""Configure YUM/DNF repositories and trust their GPG keys. DNF is a drop-in successor to YUM and reads the exact same `.repo` files under `/etc/yum.repos.d/`, so a single module covers both. If `/etc/yum.repos.d` exists, every `.repo` file already there is copied into ``, before the repositories declared in the configuration are written as new, enabled `.repo` files (so a configured repository reusing an existing file name ends up enabled, taking precedence over the copied one). A section of an existing `.repo` file is only disabled (`enabled=0`) if its `baseurl`/`mirrorlist` matches the `index_url_to_replace` pattern of one of the configured repositories; sections matching no such pattern are copied through unchanged. Each `.repo` file is written referencing its `gpg-url` (used verbatim as `gpgkey=`, exactly as YUM/DNF expect: `https://`, or a `file://` path already present on the target system). Separately, each GPG key declared in `configurations.yum.keys` is validated with `gpg` (catching a malformed key early) and written to `/etc/pki/rpm-gpg/RPM-GPG-KEY-`, the conventional location such keys live at on RPM-based systems. `gpg` is only used here to validate the key; the actual trust decision is left to YUM/DNF, which import it from that file the first time the repository is used (as `rpm --import` would). A repository's `index_url_to_replace`, if set, is an fnmatch pattern checked against the baseurl of every section of every existing `.repo` file: the repository is only written out if the pattern matches at least one of them, and skipped entirely otherwise. """ from __future__ import annotations import configparser import fnmatch import logging import shutil import subprocess import tempfile from collections.abc import Sequence from io import StringIO from id_configurator.config import NamedContent, YumRepository from id_configurator.context import Context from id_configurator.modules.base import ( Module, register, sanitize_filename, sanitize_repository_name, ) logger = logging.getLogger(__name__) _REPOS_DIR = "etc/yum.repos.d" _RPM_GPG_DIR = "etc/pki/rpm-gpg" def _disable_repo_content(content: str, patterns: Sequence[str]) -> str: parser = configparser.ConfigParser(interpolation=None) parser.optionxform = str parser.read_string(content) for section in parser.sections(): urls = [parser[section].get(key) for key in ("baseurl", "mirrorlist")] urls = [url for url in urls if url] if any(fnmatch.fnmatch(url, pattern) for url in urls for pattern in patterns): parser[section]["enabled"] = "0" buffer = StringIO() parser.write(buffer, space_around_delimiters=False) return buffer.getvalue() def _existing_index_urls(context: Context) -> set[str]: """Collect the baseurl of every section of every .repo file already present on the source filesystem.""" urls: set[str] = set() for entry in context.iter_src_dir(_REPOS_DIR): if entry.suffix != ".repo": continue parser = configparser.ConfigParser(interpolation=None) parser.optionxform = str parser.read_string(entry.read_text(encoding="utf-8")) for section in parser.sections(): # noinspection SpellCheckingInspection for key in ("baseurl", "mirrorlist"): url = parser[section].get(key) if url: urls.add(url) return urls def _disable_existing_repos(context: Context, patterns: Sequence[str]) -> None: for entry in context.iter_src_dir(_REPOS_DIR): if entry.suffix != ".repo": continue disabled = _disable_repo_content(entry.read_text(encoding="utf-8"), patterns) context.write_file(f"{_REPOS_DIR}/{entry.name}", disabled.encode("utf-8")) logger.info("copied existing yum/dnf repository file %s", entry.name) def _write_repo_file(context: Context, repository: YumRepository) -> None: name = sanitize_repository_name(repository.name) lines = [ f"[{name}]", f"name={name}", f"baseurl={repository.index_url}", "enabled=1", ] if repository.gpg_url: lines.append("gpgcheck=1") lines.append(f"gpgkey={repository.gpg_url}") else: lines.append("gpgcheck=0") context.write_file( f"{_REPOS_DIR}/{name}.repo", ("\n".join(lines) + "\n").encode("utf-8"), ) logger.info("added yum/dnf repository %s", name) def _validate_gpg_key(name: str, content: bytes) -> None: with tempfile.TemporaryDirectory() as gnupg_home: path = shutil.which("gpg") if path is not None: result = subprocess.run( [path, "--homedir", gnupg_home, "--batch", "--import"], input=content, capture_output=True, check=False, ) else: raise RuntimeError( "gpg is not available; the GNUPG package must be installed in the environment running id-configurator" ) if result.returncode != 0: raise RuntimeError( f"invalid GPG key {name}: {result.stderr.decode('utf-8', 'replace')}" ) def _import_key(context: Context, key: NamedContent) -> None: _validate_gpg_key(key.name, key.content.encode("utf-8")) context.write_file( f"{_RPM_GPG_DIR}/RPM-GPG-KEY-{sanitize_filename(key.name)}", key.content.encode("utf-8"), ) logger.info("imported yum/dnf GPG key %s", key.name) @register class YumModule(Module): name = "yum" def is_relevant(self, context: Context) -> bool: yum = context.config.configurations.yum existing_repo_files = ( entry for entry in context.iter_src_dir(_REPOS_DIR) if entry.suffix == ".repo" ) return bool(yum.repositories) or bool(yum.keys) or any(existing_repo_files) def apply(self, context: Context) -> None: patterns = [ repository.index_url_to_replace for repository in context.config.configurations.yum.repositories if repository.index_url_to_replace is not None ] _disable_existing_repos(context, patterns) existing_index_urls: set[str] | None = None for repository in context.config.configurations.yum.repositories: if repository.index_url_to_replace is not None: if existing_index_urls is None: existing_index_urls: set[str] = _existing_index_urls(context) if not any( fnmatch.fnmatch(url, repository.index_url_to_replace) for url in existing_index_urls ): logger.info( "skipped yum/dnf repository %s: index-url-to-replace %s " "matched no existing baseurl", repository.name, repository.index_url_to_replace, ) continue _write_repo_file(context, repository) for key in context.config.configurations.yum.keys: _import_key(context, key)