id_configurator/id_configurator/modules/yum.py

"""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 `<src>/etc/yum.repos.d` exists, every `.repo` file already there is
copied into `<dst>`, 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-<name>`, 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

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:
    lines = [
        f"[{repository.name}]",
        f"name={repository.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}/{sanitize_filename(repository.name)}.repo",
        ("\n".join(lines) + "\n").encode("utf-8"),
    )
    logger.info("added yum/dnf repository %s", repository.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)