id_configurator/id_configurator/modules/certificates.py

# noinspection SpellCheckingInspection
"""Install root certificates into the system's CA store.

Certificates from the configuration are dropped into the "local" CA
directory, then the trust store under `/etc/ssl/certs` (bundle + named
symlinks) is rebuilt in pure Python, mirroring what Debian's
`update-ca-certificates` shell script does: every certificate enabled in
`ca-certificates.conf` (read from `<src>`), plus everything found under the
local certificates directory, is concatenated into a single bundle file and
gets a `<name>.pem` symlink pointing at it.

The OpenSSL subject-hash symlinks (`<hash>.0`) that `update-ca-certificates`
also creates, for CApath-style directory lookups, are intentionally not
reproduced here: none of this tool's own consumers (pip, apt, dnf, apk, the
JVM, ...) do CApath lookups, they all read the single bundle file.

If `/etc/ssl/certs` is itself a symlink in `<src>` (as on RHEL, where it
points to `/etc/pki/tls/certs`), the bundle and symlinks are written to
whatever directory it resolves to instead. The write lands where the
deployed system will actually look, without ever writing a plain directory
where that system expects a symlink. In that case `/usr/lib/ssl/cert.pem`
(see below) is not written either, since it isn't OpenSSL's default CAfile
on such systems in the first place.

Otherwise, `/usr/lib/ssl/cert.pem` - OpenSSL's own compiled-in default
CAfile on Debian/Ubuntu, and what apt's own https method reads absent any
explicit configuration - is symlinked to the rebuilt bundle too. This is
normally created by the ca-certificates package's postinst script; doing it
here as well means the bundle is trusted even on a base image that never
had that package installed.

When `<src>` also has a RHEL/Fedora-style trust store (`/etc/pki/ca-trust`),
the configured certificates are additionally dropped into
`/etc/pki/ca-trust/source/anchors` and merged into
`/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem`, the single file that
`/etc/pki/tls/certs/ca-bundle.crt` and OpenSSL's own default CAfile
(`/etc/pki/tls/cert.pem`) both resolve to on such systems - which is what
dnf/curl/openssl-based tools there actually read. This intentionally does
not reproduce the rest of `update-ca-trust extract`'s output (the OpenSSL
"trust" bundle with per-certificate usage flags, the email/objsign bundles,
the extracted Java keystore, ...); only the plain PEM trust bundle is
rebuilt.
"""

from __future__ import annotations

import logging
import posixpath
from pathlib import Path

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

logger = logging.getLogger(__name__)

_LOCAL_CERTS_DIR = "usr/local/share/ca-certificates"
# noinspection SpellCheckingInspection
_CERTSCONF = "etc/ca-certificates.conf"
# noinspection SpellCheckingInspection
_CERTSDIR = "usr/share/ca-certificates"
# noinspection SpellCheckingInspection
_ETCCERTSDIR = "etc/ssl/certs"
_CERT_BUNDLE = "ca-certificates.crt"

_RHEL_ANCHORS_DIR = "etc/pki/ca-trust/source/anchors"
_RHEL_EXTRACTED_BUNDLE = "etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem"

# OpenSSL's own compiled-in default CAfile on Debian/Ubuntu (OPENSSLDIR is
# /usr/lib/ssl there); the ca-certificates package's postinst symlinks this
# to the bundle, which is what lets OpenSSL-linked tools (including apt's
# own https method) find it without any explicit CAfile configuration. That
# symlink only exists once the package has actually been installed, so it is
# created here too, making the bundle usable even on a base image that
# never had ca-certificates installed at all.
_OPENSSL_DEFAULT_CERT_FILE = "usr/lib/ssl/cert.pem"


def _pem_name(relative_crt: str) -> str:
    # noinspection GrazieInspection
    """Mirror update-ca-certificates' `.pem` symlink naming for a `*.crt` path."""
    stem = Path(relative_crt).name.removesuffix(".crt")
    stem = stem.replace(" ", "_").replace("(", "=").replace(")", "=").replace(",", "_")
    return f"{stem}.pem"


def _ensure_trailing_newline(content: bytes) -> bytes:
    return content if content.endswith(b"\n") else content + b"\n"


def _resolve_target_dir(context: Context, relative: str) -> str:
    """Resolve `relative` against `<src>`, following it if it is a symlink.

    Some systems replace a well-known directory with a symlink to another
    location (e.g. RHEL's `/etc/ssl/certs` -> `/etc/pki/tls/certs`). Writing
    to the literal path in that case would create a plain directory where
    the deployed system expects a symlink - which BuildKit refuses to
    overlay a directory onto in the first place - while leaving the actual,
    symlinked-to location untouched. Returns `relative` unchanged if it
    isn't a symlink in `<src>`, or if it can't be resolved back to a path
    under `<src>`.
    """
    if context.src is None:
        return relative
    # Not context.src_path(): it resolves symlinks away before we can even
    # check for one.
    source = context.src / relative
    if not source.is_symlink():
        return relative
    target = source.readlink()
    if target.is_absolute():
        return posixpath.normpath(target.as_posix()).lstrip("/")
    resolved = (source.parent / target).resolve()
    try:
        return resolved.relative_to(context.src.resolve()).as_posix()
    except ValueError:
        return relative


# noinspection SpellCheckingInspection
def _parse_certsconf(certsconf: Path) -> list[str]:
    """Return the paths (relative to CERTSDIR) enabled in ca-certificates.conf.

    Blank lines and comments (`#`) are ignored; lines prefixed with `!`
    are explicitly disabled, exactly like update-ca-certificates.
    """
    enabled = []
    for line in certsconf.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or line.startswith("!"):
            continue
        enabled.append(line)
    return enabled


# noinspection SpellCheckingInspection
@register
class CertificatesModule(Module):
    name = "certificates"

    def is_relevant(self, context: Context) -> bool:
        return bool(context.config.configurations.certificates)

    def apply(self, context: Context) -> None:
        configured_names = {
            sanitize_filename(certificate.name)
            for certificate in context.config.configurations.certificates
        }
        for existing in context.iter_src_dir(_LOCAL_CERTS_DIR):
            if existing.stem not in configured_names:
                context.write_file(
                    f"{_LOCAL_CERTS_DIR}/{existing.name}", existing.read_bytes()
                )
        for certificate in context.config.configurations.certificates:
            context.write_file(
                f"{_LOCAL_CERTS_DIR}/{sanitize_filename(certificate.name)}.crt",
                certificate.content.encode(),
            )
            logger.info("installed root certificate %s", certificate.name)
        self._update_ca_certificates_conf(context)
        if self._rhel_trust_store_present(context):
            self._update_rhel_trust_store(context, configured_names)

    @staticmethod
    def _rhel_trust_store_present(context: Context) -> bool:
        anchors = context.src_path(_RHEL_ANCHORS_DIR)
        return anchors is not None and anchors.is_dir()

    @staticmethod
    def _collect_conf_entries(context: Context) -> list[tuple[str, bytes]]:
        certsconf = context.src_path(_CERTSCONF)
        certsdir = context.src_path(_CERTSDIR)
        if certsconf is None or not certsconf.is_file() or certsdir is None:
            return []
        entries = []
        for relative in _parse_certsconf(certsconf):
            source = certsdir / relative
            if not source.is_file():
                logger.warning("%s not found, but listed in %s", source, certsconf)
                continue
            entries.append((f"{_CERTSDIR}/{relative}", source.read_bytes()))
        return entries

    @staticmethod
    def _collect_local_entries(localcertsdir: Path) -> list[tuple[str, bytes]]:
        entries = []
        for source in sorted(localcertsdir.rglob("*.crt")):
            relative = source.relative_to(localcertsdir).as_posix()
            entries.append((f"{_LOCAL_CERTS_DIR}/{relative}", source.read_bytes()))
        return entries

    @classmethod
    def _update_ca_certificates_conf(cls, context: Context) -> None:
        etccertsdir_relative = _resolve_target_dir(context, _ETCCERTSDIR)
        etccertsdir = context.dst_path(etccertsdir_relative)
        localcertsdir = context.dst_path(_LOCAL_CERTS_DIR)
        etccertsdir.mkdir(parents=True, exist_ok=True)
        localcertsdir.mkdir(parents=True, exist_ok=True)

        # Same processing order as update-ca-certificates: ca-certificates.conf
        # entries first, then everything under the local certificate directory.
        entries = cls._collect_conf_entries(context) + cls._collect_local_entries(
            localcertsdir
        )

        bundle = bytearray()
        for relative, content in entries:
            bundle += _ensure_trailing_newline(content)
            pem_path = etccertsdir / _pem_name(relative)
            pem_path.unlink(missing_ok=True)
            pem_path.symlink_to(f"/{relative}")

        context.write_file(f"{etccertsdir_relative}/{_CERT_BUNDLE}", bytes(bundle))
        logger.info("rebuilt %s from %d certificate(s)", _CERT_BUNDLE, len(entries))

        if etccertsdir_relative == _ETCCERTSDIR:
            # Not context.dst_path(_OPENSSL_DEFAULT_CERT_FILE) directly: on a
            # host that already has this symlink (e.g. re-applying to a live
            # `--dst /`), resolving the leaf itself would follow it straight
            # to the bundle we just wrote, and the unlink+symlink_to below
            # would then target the bundle's own path instead of this one -
            # replacing the bundle with a symlink to itself. Resolving only
            # the containing directory keeps the leaf name literal.
            cert_file_dir = context.dst_path(
                posixpath.dirname(_OPENSSL_DEFAULT_CERT_FILE)
            )
            cert_file_dir.mkdir(parents=True, exist_ok=True)
            cert_file = cert_file_dir / posixpath.basename(_OPENSSL_DEFAULT_CERT_FILE)
            cert_file.unlink(missing_ok=True)
            cert_file.symlink_to(f"/{etccertsdir_relative}/{_CERT_BUNDLE}")

    @staticmethod
    def _update_rhel_trust_store(context: Context, configured_names: set[str]) -> None:
        for existing in context.iter_src_dir(_RHEL_ANCHORS_DIR):
            if existing.stem not in configured_names:
                context.write_file(
                    f"{_RHEL_ANCHORS_DIR}/{existing.name}", existing.read_bytes()
                )
        for certificate in context.config.configurations.certificates:
            context.write_file(
                f"{_RHEL_ANCHORS_DIR}/{sanitize_filename(certificate.name)}.crt",
                certificate.content.encode(),
            )

        bundle = bytearray()
        existing_bundle = context.read_src_file(_RHEL_EXTRACTED_BUNDLE)
        if existing_bundle is not None:
            bundle += _ensure_trailing_newline(existing_bundle)

        # Every write above went through context.write_file(), which only ever
        # creates regular files, so every entry here is a certificate to add.
        anchors = sorted(context.dst_path(_RHEL_ANCHORS_DIR).iterdir())
        for anchor in anchors:
            bundle += _ensure_trailing_newline(anchor.read_bytes())

        context.write_file(_RHEL_EXTRACTED_BUNDLE, bytes(bundle))
        logger.info("rebuilt RHEL/Fedora trust store from %d anchor(s)", len(anchors))