"""Detect existing APT sources and configure new ones, in the DEB822 format. If `/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 `/etc/apt`, and the sources declared in the configuration are added as new, enabled `.sources` files. An existing entry (a single `deb`/`deb-src` line, or a DEB822 stanza) is only disabled if its URI matches the `uri_to_replace` pattern of one of the configured sources; entries matching no such pattern are copied through unchanged, still enabled. Each source can be trusted two ways: `signed_by` holds the ASCII-armored content of a GPG public key, folded inline as a multi-line `Signed-By` field (per sources.list(5) - verified empirically against a real, genuinely signed repository, so no `gpg` binary or separate keyring file is needed at all to apply this configuration). `signed_by_file` instead names a keyring already expected to exist on the target system, used verbatim as `Signed-By`; if both are given, `signed_by` takes precedence. A source's `uri_to_replace`, if set, is an fnmatch pattern checked against the URIs of the sources already present on the source filesystem: the source is only written out if the pattern matches at least one of them, and skipped entirely otherwise - even on an otherwise Debian-like machine. """ from __future__ import annotations import fnmatch 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, sanitize_filename logger = logging.getLogger(__name__) _APT_DIR = "etc/apt" _SOURCES_LIST = "etc/apt/sources.list" _SOURCES_LIST_D = "etc/apt/sources.list.d" def _uri_from_deb_line(stripped_line: str) -> str | None: """Extract the URI from a single, already-stripped `deb`/`deb-src` line (`deb [options] uri suite [component...]`), or `None` if the line is not such a line.""" parts = stripped_line.split() if not parts or parts[0] not in ("deb", "deb-src"): return None parts = parts[1:] while parts and parts[0].startswith("["): while parts and not parts[0].endswith("]"): parts.pop(0) if parts: parts.pop(0) return parts[0] if parts else None def _disable_list_content(content: str, patterns: Sequence[str]) -> str: lines = [] for line in content.splitlines(): stripped = line.strip() uri = ( None if not stripped or stripped.startswith("#") else _uri_from_deb_line(stripped) ) if uri is not None and any( fnmatch.fnmatch(uri, pattern) for pattern in patterns ): lines.append(f"# {line}") else: lines.append(line) return "\n".join(lines) + "\n" def _uris_from_deb822_stanza(stanza: str) -> set[str]: uris: set[str] = set() for line in stanza.splitlines(): if line.lower().startswith("uris:"): uris.update(line.split(":", 1)[1].split()) return uris def _disable_deb822_content(content: str, patterns: Sequence[str]) -> str: stanzas = [] for stanza in content.split("\n\n"): if not stanza.strip(): continue lines = stanza.splitlines() uris = _uris_from_deb822_stanza(stanza) if any(fnmatch.fnmatch(uri, pattern) for uri in uris for pattern in patterns): lines = [line for line in lines if not line.lower().startswith("enabled:")] lines.append("Enabled: no") stanzas.append("\n".join(lines)) return "\n\n".join(stanzas) + "\n" def _uris_from_list_content(content: str) -> set[str]: """Extract the URIs declared in a classic one-line `.list` file (`deb [options] uri suite [component...]`), skipping comments and any bracketed option list.""" uris: set[str] = set() for line in content.splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue uri = _uri_from_deb_line(stripped) if uri is not None: uris.add(uri) return uris def _uris_from_deb822_content(content: str) -> set[str]: """Extract the URIs declared in the `URIs:` field of each stanza of a DEB822 `.sources` file.""" uris: set[str] = set() for stanza in content.split("\n\n"): uris.update(_uris_from_deb822_stanza(stanza)) return uris def _existing_uris(context: Context) -> set[str]: """Collect every URI declared by the apt sources already present on the source filesystem, whether in the top-level sources.list or under sources.list.d/, in either the classic or the DEB822 format.""" uris: set[str] = set() existing_top_level = context.read_src_file(_SOURCES_LIST) if existing_top_level is not None: uris.update(_uris_from_list_content(existing_top_level.decode("utf-8"))) for entry in context.iter_src_dir(_SOURCES_LIST_D): if entry.suffix == ".sources": uris.update(_uris_from_deb822_content(entry.read_text(encoding="utf-8"))) elif entry.suffix == ".list": uris.update(_uris_from_list_content(entry.read_text(encoding="utf-8"))) return uris @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: patterns = [ source.uri_to_replace for source in context.config.configurations.apt.sources if source.uri_to_replace is not None ] self._copy_existing_disabled(context, patterns) self._write_configured_sources(context) @staticmethod def _copy_existing_disabled(context: Context, patterns: Sequence[str]) -> 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"), patterns ) context.write_file(_SOURCES_LIST, disabled.encode("utf-8")) logger.info("copied 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"), patterns ) elif entry.suffix == ".list": disabled = _disable_list_content( entry.read_text(encoding="utf-8"), patterns ) else: continue context.write_file( f"{_SOURCES_LIST_D}/{entry.name}", disabled.encode("utf-8") ) logger.info("copied existing apt source %s", entry.name) @staticmethod def _write_configured_sources(context: Context) -> None: existing_uris: set[str] | None = None for source in context.config.configurations.apt.sources: if source.uri_to_replace is not None: if existing_uris is None: existing_uris = _existing_uris(context) if not any( fnmatch.fnmatch(uri, source.uri_to_replace) for uri in existing_uris ): logger.info( "skipped apt source %s: uri-to-replace %s matched no " "existing uri", source.name, source.uri_to_replace, ) continue context.write_file( f"{_SOURCES_LIST_D}/{sanitize_filename(source.name)}.sources", source.to_deb822( signed_by_armored_key=source.signed_by, signed_by_path=source.signed_by_file, ).encode("utf-8"), ) logger.info("added apt source %s", source.name)