id_configurator/id_configurator/modules/apt.py

"""Detect existing APT sources and configure new ones, in the DEB822 format.

If `<src>/etc/apt` exists, any source file already present there (classic
one-line `.list` files or DEB822 `.sources` files, whether at the top level
or under `sources.list.d/`) is copied into `<dst>/etc/apt`, disabled so it no
longer takes effect, and the sources declared in the configuration are added
as new, enabled `.sources` files.

Each source can be trusted two ways: `signed_by` holds the ASCII-armored
content of a GPG public key, which is converted with `gpg --dearmor` into a
binary keyring dedicated to this one source, written to
`/etc/apt/keyrings/<name>.gpg` - it is that path, not the key's own content,
that ends up in the rendered `.sources` file's `Signed-By` field.
`signed_by_file` instead names a keyring already expected to exist on the
target system, used verbatim as `Signed-By` (no file is written for it); if
both are given, `signed_by` takes precedence. `gpg` is only used here for
the armored-to-binary format conversion; the actual signature verification
of the repository itself is left to APT when it fetches the repository's
Release/InRelease file.
"""

from __future__ import annotations

import logging
import shutil
import subprocess
import tempfile
from pathlib import Path

from id_configurator.config import AptSource
from id_configurator.context import Context
from id_configurator.modules.base import Module, register

logger = logging.getLogger(__name__)

_APT_DIR = "etc/apt"
_SOURCES_LIST = "etc/apt/sources.list"
_SOURCES_LIST_D = "etc/apt/sources.list.d"
_KEYRINGS_DIR = "etc/apt/keyrings"


def _disable_list_content(content: str) -> str:
    lines = [
        line if not line.strip() or line.strip().startswith("#") else f"# {line}"
        for line in content.splitlines()
    ]
    return "\n".join(lines) + "\n"


def _disable_deb822_content(content: str) -> str:
    disabled_stanzas = []
    for stanza in content.split("\n\n"):
        if not stanza.strip():
            continue
        lines = stanza.splitlines()
        new_lines = [line for line in lines if not line.lower().startswith("enabled:")]
        new_lines.append("Enabled: no")
        disabled_stanzas.append("\n".join(new_lines))
    return "\n\n".join(disabled_stanzas) + "\n"


def _dearmor_gpg_key(name: str, content: bytes) -> bytes:
    """Convert an ASCII-armored GPG public key into apt's binary keyring format."""
    with tempfile.TemporaryDirectory() as gnupg_home:
        path = shutil.which("gpg")
        if path is not None:
            result = subprocess.run(
                [path, "--homedir", gnupg_home, "--batch", "--dearmor"],
                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 for apt source {name}: {result.stderr.decode('utf-8', 'replace')}"
        )
    return result.stdout


@register
class AptModule(Module):
    name = "apt"

    def is_relevant(self, context: Context) -> bool:
        return self._src_apt_dir(context) is not None

    @staticmethod
    def _src_apt_dir(context: Context) -> Path | None:
        path = context.src_path(_APT_DIR)
        if path is not None and path.is_dir():
            return path
        return None

    def apply(self, context: Context) -> None:
        self._copy_existing_disabled(context)
        self._write_configured_sources(context)

    @staticmethod
    def _copy_existing_disabled(context: Context) -> None:
        existing_top_level = context.read_src_file(_SOURCES_LIST)
        if existing_top_level is not None:
            disabled = _disable_list_content(existing_top_level.decode("utf-8"))
            context.write_file(_SOURCES_LIST, disabled.encode("utf-8"))
            logger.info("disabled existing apt source list %s", _SOURCES_LIST)

        for entry in context.iter_src_dir(_SOURCES_LIST_D):
            if entry.suffix == ".sources":
                disabled = _disable_deb822_content(entry.read_text(encoding="utf-8"))
            elif entry.suffix == ".list":
                disabled = _disable_list_content(entry.read_text(encoding="utf-8"))
            else:
                continue
            context.write_file(
                f"{_SOURCES_LIST_D}/{entry.name}", disabled.encode("utf-8")
            )
            logger.info("disabled existing apt source %s", entry.name)

    def _resolve_signed_by(self, context: Context, source: AptSource) -> str | None:
        if source.signed_by:
            keyring = _dearmor_gpg_key(source.name, source.signed_by.encode("utf-8"))
            relative = f"{_KEYRINGS_DIR}/{source.name}.gpg"
            context.write_file(relative, keyring)
            logger.info("installed apt keyring for source %s", source.name)
            return f"/{relative}"
        if source.signed_by_file:
            return source.signed_by_file
        return None

    def _write_configured_sources(self, context: Context) -> None:
        for source in context.config.configurations.apt.sources:
            signed_by_path = self._resolve_signed_by(context, source)
            context.write_file(
                f"{_SOURCES_LIST_D}/{source.name}.sources",
                source.to_deb822(signed_by_path).encode("utf-8"),
            )
            logger.info("added apt source %s", source.name)