feat: initial commit

id_configurator/modules/apt.py

@@ -0,0 +1,100 @@
+"""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.
+"""
+
+from __future__ import annotations
+
+import base64
+import logging
+from pathlib import Path
+
+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"
+
+
+@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_keys(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)
+
+    @staticmethod
+    def _write_configured_keys(context: Context) -> None:
+        for key in context.config.configurations.apt.keys:
+            content = base64.b64decode(key.content)
+            context.write_file(f"{_KEYRINGS_DIR}/{key.name}.asc", content)
+            logger.info("installed apt keyring %s", key.name)
+
+    @staticmethod
+    def _write_configured_sources(context: Context) -> None:
+        for source in context.config.configurations.apt.sources:
+            context.write_file(
+                f"{_SOURCES_LIST_D}/{source.name}.sources",
+                source.to_deb822().encode("utf-8"),
+            )
+            logger.info("added apt source %s", source.name)