id_configurator/id_configurator/modules/apk.py

"""Detect existing APK (Alpine) repositories and configure new ones.

If `<src>/etc/apk` exists, the existing `/etc/apk/repositories` file (if any)
is copied into `<dst>`, and the repositories declared in the configuration
are appended as new, active lines. An existing active line is only disabled
(commented out) if it matches the `url_to_replace` pattern of one of the
configured repositories; lines matching no such pattern are copied through
unchanged. Public keys provided alongside a repository are written to
`<dst>/etc/apk/keys/`, named after their content so that repositories
sharing the same key (a common case for official Alpine mirrors) only
produce a single key file.

A repository's `url_to_replace`, if set, is a fnmatch pattern checked
against the URLs already active in /etc/apk/repositories: the repository
is only written out if the pattern matches at least one of them, and
skipped entirely otherwise.
"""

from __future__ import annotations

import fnmatch
import hashlib
import logging
from collections.abc import Sequence
from pathlib import Path

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

logger = logging.getLogger(__name__)

_APK_DIR = "etc/apk"
_REPOSITORIES = "etc/apk/repositories"
_KEYS_DIR = "etc/apk/keys"


def _disable_repositories_content(content: str, patterns: Sequence[str]) -> str:
    lines = []
    for line in content.splitlines():
        stripped = line.strip()
        if (
            stripped
            and not stripped.startswith("#")
            and any(fnmatch.fnmatch(stripped, pattern) for pattern in patterns)
        ):
            lines.append(f"# {line}")
        else:
            lines.append(line)
    return "\n".join(lines) + "\n"


def _key_filename(content: bytes) -> str:
    return f"{hashlib.sha256(content).hexdigest()[:16]}.rsa.pub"


def _is_valid_key_filename(name: str) -> bool:
    """Check that `name` is safe to use as a single filename under `_KEYS_DIR`.

    Unlike `sanitize_filename`, an invalid name is not cleaned up: the
    caller falls back to the content-derived name instead, since a
    configured `key_filename` that isn't a plain file name (e.g. containing
    `/` or `\\`, which would turn into extra path segments, or `.`/`..`)
    most likely indicates a configuration mistake.
    """
    return name not in ("", ".", "..") and "/" not in name and "\\" not in name


def _existing_urls(context: Context) -> set[str]:
    """Collect the URLs already declared in /etc/apk/repositories on the
    source filesystem (one active, non-comment line per URL)."""
    existing = context.read_src_file(_REPOSITORIES)
    if existing is None:
        return set()
    return {
        stripped
        for line in existing.decode("utf-8").splitlines()
        if (stripped := line.strip()) and not stripped.startswith("#")
    }


@register
class ApkModule(Module):
    name = "apk"

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

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

    def apply(self, context: Context) -> None:
        patterns = [
            repository.url_to_replace
            for repository in context.config.configurations.apk.repositories
            if repository.url_to_replace is not None
        ]
        disabled = self._disabled_existing_repositories(context, patterns)
        configured = self._configured_repositories_lines(context)
        content = disabled + ("\n".join(configured) + "\n" if configured else "")
        if content:
            context.write_file(_REPOSITORIES, content.encode("utf-8"))

    @staticmethod
    def _disabled_existing_repositories(
        context: Context, patterns: Sequence[str]
    ) -> str:
        existing = context.read_src_file(_REPOSITORIES)
        if existing is None:
            return ""
        logger.info("copied existing apk repositories %s", _REPOSITORIES)
        return _disable_repositories_content(existing.decode("utf-8"), patterns)

    @staticmethod
    def _configured_repositories_lines(context: Context) -> list[str]:
        lines = []
        existing_urls: set[str] | None = None
        for repository in context.config.configurations.apk.repositories:
            if repository.url_to_replace is not None:
                if existing_urls is None:
                    existing_urls: set[str] = _existing_urls(context)
                if not any(
                    fnmatch.fnmatch(url, repository.url_to_replace)
                    for url in existing_urls
                ):
                    logger.info(
                        "skipped apk repository %s: url-to-replace %s matched "
                        "no existing url",
                        repository.name,
                        repository.url_to_replace,
                    )
                    continue
            lines.append(repository.url)
            logger.info("added apk repository %s", repository.name)
            if repository.key:
                if repository.key_filename and _is_valid_key_filename(
                    repository.key_filename
                ):
                    filename = repository.key_filename
                else:
                    filename = _key_filename(repository.key.encode("utf-8"))
                context.write_file(
                    f"{_KEYS_DIR}/{filename}", repository.key.encode("utf-8")
                )
                logger.info("installed apk key %s", filename)
        return lines