feat: enhance repository configuration to support url_to_replace patterns for disabling existing entries

Signed-off-by: d9pouces <github@19pouces.net>
This commit is contained in:
d9pouces
2026-09-19 23:40:35 +02:00
parent b2dcae0daf
commit cea45c97ff
8 changed files with 757 additions and 66 deletions
+133 -21
View File
@@ -2,9 +2,13 @@
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.
or under `sources.list.d/`) is copied into `<dst>/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`
@@ -13,11 +17,18 @@ 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
@@ -30,24 +41,100 @@ _SOURCES_LIST = "etc/apt/sources.list"
_SOURCES_LIST_D = "etc/apt/sources.list.d"
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()
]
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 _disable_deb822_content(content: str) -> str:
disabled_stanzas = []
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()
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"
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
@@ -65,32 +152,57 @@ class AptModule(Module):
return None
def apply(self, context: Context) -> None:
self._copy_existing_disabled(context)
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) -> None:
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"))
disabled = _disable_list_content(
existing_top_level.decode("utf-8"), patterns
)
context.write_file(_SOURCES_LIST, disabled.encode("utf-8"))
logger.info("disabled existing apt source list %s", _SOURCES_LIST)
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"))
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"))
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("disabled existing apt source %s", entry.name)
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(