diff --git a/id_configurator/config.py b/id_configurator/config.py index c964669..aed9f19 100644 --- a/id_configurator/config.py +++ b/id_configurator/config.py @@ -46,6 +46,10 @@ class AptSource(BaseModel): - `signed_by_file`: the *path* to a keyring already expected to exist on the target system, written to the rendered `Signed-By` field verbatim. If both are set, `signed_by` takes precedence. + + `uri_to_replace`, if set, is an fnmatch pattern: this source is only + applied if it matches at least one URI already present in the machine's + existing apt configuration; otherwise it is skipped entirely. """ model_config = ConfigDict(populate_by_name=True) @@ -75,6 +79,7 @@ class AptSource(BaseModel): check_date: bool | None = Field(default=None, alias="check-date") date_max_future: str | None = Field(default=None, alias="date-max-future") description: str | None = None + uri_to_replace: str | None = Field(default=None, alias="uri-to-replace") def to_deb822( self, @@ -143,11 +148,17 @@ class AptConfig(BaseModel): class ApkRepository(BaseModel): - """An Alpine APK repository (see /etc/apk/repositories), e.g. a URL ending in /main or /community.""" + """An Alpine APK repository (see /etc/apk/repositories), e.g. a URL ending in /main or /community. + + `url_to_replace`, if set, is an fnmatch pattern: this repository is only + applied if it matches at least one URL already present in the machine's + existing apk configuration; otherwise it is skipped entirely. + """ name: str url: str key: str | None = None + url_to_replace: str | None = None class ApkConfig(BaseModel): @@ -155,10 +166,19 @@ class ApkConfig(BaseModel): class YumRepository(BaseModel): + """A YUM/DNF repository (see /etc/yum.repos.d/*.repo). + + `index_url_to_replace`, if set, is an fnmatch pattern: this repository is + only applied if it matches at least one baseurl already present in the + machine's existing yum/dnf configuration; otherwise it is skipped + entirely. + """ + model_config = ConfigDict(populate_by_name=True) name: str index_url: str = Field(alias="index-url") gpg_url: str | None = Field(default=None, alias="gpg-url") + index_url_to_replace: str | None = Field(default=None, alias="index-url-to-replace") class YumConfig(BaseModel): diff --git a/id_configurator/modules/apk.py b/id_configurator/modules/apk.py index 2469db9..3cd2092 100644 --- a/id_configurator/modules/apk.py +++ b/id_configurator/modules/apk.py @@ -1,18 +1,27 @@ """Detect existing APK (Alpine) repositories and configure new ones. If `/etc/apk` exists, the existing `/etc/apk/repositories` file (if any) -is copied into `` with every active line disabled (commented out), and -the repositories declared in the configuration are appended as new, active -lines. Public keys provided alongside a repository are written to +is copied into ``, 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 `/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 an 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 @@ -25,11 +34,18 @@ _REPOSITORIES = "etc/apk/repositories" _KEYS_DIR = "etc/apk/keys" -def _disable_repositories_content(content: str) -> str: - lines = [ - line if not line.strip() or line.strip().startswith("#") else f"# {line}" - for line in content.splitlines() - ] +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" @@ -37,6 +53,19 @@ def _key_filename(content: bytes) -> str: return f"{hashlib.sha256(content).hexdigest()[:16]}.rsa.pub" +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" @@ -52,24 +81,46 @@ class ApkModule(Module): return None def apply(self, context: Context) -> None: - disabled = self._disabled_existing_repositories(context) + 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) -> str: + def _disabled_existing_repositories( + context: Context, patterns: Sequence[str] + ) -> str: existing = context.read_src_file(_REPOSITORIES) if existing is None: return "" - logger.info("disabled existing apk repositories %s", _REPOSITORIES) - return _disable_repositories_content(existing.decode("utf-8")) + 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 = _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: diff --git a/id_configurator/modules/apt.py b/id_configurator/modules/apt.py index 765b0d9..8f6fb0e 100644 --- a/id_configurator/modules/apt.py +++ b/id_configurator/modules/apt.py @@ -2,9 +2,13 @@ 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`, 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 `/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( diff --git a/id_configurator/modules/yum.py b/id_configurator/modules/yum.py index d0caa7a..0cd70e3 100644 --- a/id_configurator/modules/yum.py +++ b/id_configurator/modules/yum.py @@ -4,10 +4,13 @@ DNF is a drop-in successor to YUM and reads the exact same `.repo` files under `/etc/yum.repos.d/`, so a single module covers both. If `/etc/yum.repos.d` exists, every `.repo` file already there is -copied into `` with all of its sections disabled (`enabled=0`), before -the repositories declared in the configuration are written as new, enabled -`.repo` files (so a configured repository reusing an existing file name -ends up enabled, taking precedence over the disabled copy). Each `.repo` +copied into ``, before the repositories declared in the configuration +are written as new, enabled `.repo` files (so a configured repository +reusing an existing file name ends up enabled, taking precedence over the +copied one). A section of an existing `.repo` file is only disabled +(`enabled=0`) if its `baseurl`/`mirrorlist` matches the +`index_url_to_replace` pattern of one of the configured repositories; +sections matching no such pattern are copied through unchanged. Each `.repo` file is written referencing its `gpg-url` (used verbatim as `gpgkey=`, exactly as YUM/DNF expect: `https://`, or a `file://` path already present on the target system). Separately, each GPG key declared in @@ -17,15 +20,22 @@ conventional location such keys live at on RPM-based systems. `gpg` is only used here to validate the key; the actual trust decision is left to YUM/DNF, which import it from that file the first time the repository is used (as `rpm --import` would). + +A repository's `index_url_to_replace`, if set, is an fnmatch pattern +checked against the baseurl of every section of every existing `.repo` +file: the repository is only written out if the pattern matches at least +one of them, and skipped entirely otherwise. """ from __future__ import annotations import configparser +import fnmatch import logging import shutil import subprocess import tempfile +from collections.abc import Sequence from io import StringIO from id_configurator.config import NamedContent, YumRepository @@ -38,24 +48,46 @@ _REPOS_DIR = "etc/yum.repos.d" _RPM_GPG_DIR = "etc/pki/rpm-gpg" -def _disable_repo_content(content: str) -> str: +def _disable_repo_content(content: str, patterns: Sequence[str]) -> str: parser = configparser.ConfigParser(interpolation=None) parser.optionxform = str parser.read_string(content) for section in parser.sections(): - parser[section]["enabled"] = "0" + urls = [parser[section].get(key) for key in ("baseurl", "mirrorlist")] + urls = [url for url in urls if url] + if any(fnmatch.fnmatch(url, pattern) for url in urls for pattern in patterns): + parser[section]["enabled"] = "0" buffer = StringIO() parser.write(buffer, space_around_delimiters=False) return buffer.getvalue() -def _disable_existing_repos(context: Context) -> None: +def _existing_index_urls(context: Context) -> set[str]: + """Collect the baseurl of every section of every .repo file already + present on the source filesystem.""" + urls: set[str] = set() for entry in context.iter_src_dir(_REPOS_DIR): if entry.suffix != ".repo": continue - disabled = _disable_repo_content(entry.read_text(encoding="utf-8")) + parser = configparser.ConfigParser(interpolation=None) + parser.optionxform = str + parser.read_string(entry.read_text(encoding="utf-8")) + for section in parser.sections(): + # noinspection SpellCheckingInspection + for key in ("baseurl", "mirrorlist"): + url = parser[section].get(key) + if url: + urls.add(url) + return urls + + +def _disable_existing_repos(context: Context, patterns: Sequence[str]) -> None: + for entry in context.iter_src_dir(_REPOS_DIR): + if entry.suffix != ".repo": + continue + disabled = _disable_repo_content(entry.read_text(encoding="utf-8"), patterns) context.write_file(f"{_REPOS_DIR}/{entry.name}", disabled.encode("utf-8")) - logger.info("disabled existing yum/dnf repository file %s", entry.name) + logger.info("copied existing yum/dnf repository file %s", entry.name) def _write_repo_file(context: Context, repository: YumRepository) -> None: @@ -120,8 +152,28 @@ class YumModule(Module): return bool(yum.repositories) or bool(yum.keys) or any(existing_repo_files) def apply(self, context: Context) -> None: - _disable_existing_repos(context) + patterns = [ + repository.index_url_to_replace + for repository in context.config.configurations.yum.repositories + if repository.index_url_to_replace is not None + ] + _disable_existing_repos(context, patterns) + existing_index_urls: set[str] | None = None for repository in context.config.configurations.yum.repositories: + if repository.index_url_to_replace is not None: + if existing_index_urls is None: + existing_index_urls: set[str] = _existing_index_urls(context) + if not any( + fnmatch.fnmatch(url, repository.index_url_to_replace) + for url in existing_index_urls + ): + logger.info( + "skipped yum/dnf repository %s: index-url-to-replace %s " + "matched no existing baseurl", + repository.name, + repository.index_url_to_replace, + ) + continue _write_repo_file(context, repository) for key in context.config.configurations.yum.keys: _import_key(context, key) diff --git a/library/id_configurator.py b/library/id_configurator.py index bc93ff2..d3c7d73 100644 --- a/library/id_configurator.py +++ b/library/id_configurator.py @@ -167,6 +167,13 @@ options: description: type: str description: DEB822 "Description" field. + uri_to_replace: + type: str + description: + - fnmatch pattern checked against the URIs already present in + the machine's existing apt configuration. If set, this + source is only applied when the pattern matches at least + one of them; otherwise it is skipped entirely. apk: type: dict default: {} @@ -185,6 +192,13 @@ options: key: type: str description: The signing key's name, if any (see NamedContent). + url_to_replace: + type: str + description: + - fnmatch pattern checked against the URLs already active in + the machine's existing /etc/apk/repositories. If set, this + repository is only applied when the pattern matches at + least one of them; otherwise it is skipped entirely. yum: type: dict default: {} @@ -207,6 +221,13 @@ options: gpg_url: type: str description: The repository's gpgkey URL. + index_url_to_replace: + type: str + description: + - fnmatch pattern checked against the baseurl of every + section of every existing .repo file. If set, this + repository is only applied when the pattern matches at + least one of them; otherwise it is skipped entirely. keys: type: list elements: dict @@ -491,6 +512,7 @@ ARGUMENT_SPEC = { "check_date": {"type": "bool"}, "date_max_future": {"type": "str"}, "description": {"type": "str"}, + "uri_to_replace": {"type": "str"}, }, }, }, @@ -506,6 +528,7 @@ ARGUMENT_SPEC = { "options": { "url": {"type": "str", "required": True}, "key": {"type": "str"}, + "url_to_replace": {"type": "str"}, }, }, }, @@ -522,6 +545,7 @@ ARGUMENT_SPEC = { "name": {"type": "str", "required": True}, "index_url": {"type": "str", "required": True}, "gpg_url": {"type": "str"}, + "index_url_to_replace": {"type": "str"}, }, }, "keys": { diff --git a/tests/test_apk.py b/tests/test_apk.py index 2a191cf..4adecc9 100644 --- a/tests/test_apk.py +++ b/tests/test_apk.py @@ -48,7 +48,9 @@ def test_is_relevant_true_with_apk_dir(tmp_path) -> None: assert ApkModule().is_relevant(context) is True -def test_apply_disables_existing_and_adds_configured_repositories(tmp_path) -> None: +def test_apply_copies_existing_repositories_unchanged_without_url_to_replace( + tmp_path, +) -> None: src = tmp_path / "src" apk_dir = src / "etc/apk" apk_dir.mkdir(parents=True) @@ -62,11 +64,13 @@ def test_apply_disables_existing_and_adds_configured_repositories(tmp_path) -> N ApkModule().apply(context) + # No configured repository declares a url_to_replace, so the existing + # lines are copied through unchanged, still active. content = (dst / "etc/apk/repositories").read_text(encoding="utf-8") assert content == ( "# comment\n" - "# https://old-mirror.invalid/v3.18/main\n" - "# https://old-mirror.invalid/v3.18/community\n" + "https://old-mirror.invalid/v3.18/main\n" + "https://old-mirror.invalid/v3.18/community\n" "https://example.invalid/v3.19/main\n" "https://example.invalid/v3.19/community\n" ) @@ -75,6 +79,42 @@ def test_apply_disables_existing_and_adds_configured_repositories(tmp_path) -> N assert key_content == KEY_BYTES +def test_apply_disables_only_existing_lines_matching_url_to_replace(tmp_path) -> None: + src = tmp_path / "src" + apk_dir = src / "etc/apk" + apk_dir.mkdir(parents=True) + (apk_dir / "repositories").write_text( + "https://old-mirror.invalid/v3.18/main\nhttps://old-mirror.invalid/v3.18/community\n", + encoding="utf-8", + ) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "apk": { + "repositories": [ + { + "name": "main", + "url": "https://example.invalid/v3.19/main", + "url_to_replace": "https://old-mirror.invalid/v3.18/main", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=Configuration.model_validate(data), dst=dst, src=src) + + ApkModule().apply(context) + + content = (dst / "etc/apk/repositories").read_text(encoding="utf-8") + assert content == ( + "# https://old-mirror.invalid/v3.18/main\n" + "https://old-mirror.invalid/v3.18/community\n" + "https://example.invalid/v3.19/main\n" + ) + + def test_apply_without_existing_repositories_file(tmp_path) -> None: src = tmp_path / "src" (src / "etc/apk").mkdir(parents=True) @@ -90,6 +130,99 @@ def test_apply_without_existing_repositories_file(tmp_path) -> None: ) +def test_apply_skips_repository_when_url_to_replace_matches_nothing(tmp_path) -> None: + src = tmp_path / "src" + apk_dir = src / "etc/apk" + apk_dir.mkdir(parents=True) + (apk_dir / "repositories").write_text( + "https://old-mirror.invalid/v3.18/main\n", encoding="utf-8" + ) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "apk": { + "repositories": [ + { + "name": "main", + "url": "https://example.invalid/v3.19/main", + "url_to_replace": "https://other.invalid/*", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=Configuration.model_validate(data), dst=dst, src=src) + + ApkModule().apply(context) + + content = (dst / "etc/apk/repositories").read_text(encoding="utf-8") + assert "example.invalid" not in content + + +def test_apply_writes_repository_when_url_to_replace_matches_existing(tmp_path) -> None: + src = tmp_path / "src" + apk_dir = src / "etc/apk" + apk_dir.mkdir(parents=True) + (apk_dir / "repositories").write_text( + "https://old-mirror.invalid/v3.18/main\n", encoding="utf-8" + ) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "apk": { + "repositories": [ + { + "name": "main", + "url": "https://example.invalid/v3.19/main", + "url_to_replace": "https://old-mirror.invalid/*", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=Configuration.model_validate(data), dst=dst, src=src) + + ApkModule().apply(context) + + content = (dst / "etc/apk/repositories").read_text(encoding="utf-8") + assert "https://example.invalid/v3.19/main" in content + + +def test_apply_reuses_cached_existing_urls_and_handles_missing_file(tmp_path) -> None: + src = tmp_path / "src" + (src / "etc/apk").mkdir(parents=True) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "apk": { + "repositories": [ + { + "name": "main", + "url": "https://example.invalid/v3.19/main", + "url_to_replace": "https://old.invalid/*", + }, + { + "name": "community", + "url": "https://example.invalid/v3.19/community", + "url_to_replace": "https://old.invalid/*", + }, + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=Configuration.model_validate(data), dst=dst, src=src) + + ApkModule().apply(context) + + assert not (dst / "etc/apk/repositories").exists() + + def test_apply_writes_nothing_when_no_repositories_at_all(tmp_path) -> None: empty_config = Configuration.model_validate( {"uuid": "12345678-1234-5678-1234-567812345678", "organisation": "InterDiode"} diff --git a/tests/test_apt.py b/tests/test_apt.py index 5accb33..4c9a348 100644 --- a/tests/test_apt.py +++ b/tests/test_apt.py @@ -70,7 +70,9 @@ def test_is_relevant_true_with_apt_dir(tmp_path) -> None: assert AptModule().is_relevant(context) is True -def test_apply_disables_existing_sources_and_adds_configured_ones(tmp_path) -> None: +def test_apply_copies_existing_sources_unchanged_without_uri_to_replace( + tmp_path, +) -> None: src = tmp_path / "src" apt_dir = src / "etc/apt" sources_list_d = apt_dir / "sources.list.d" @@ -97,24 +99,24 @@ def test_apply_disables_existing_sources_and_adds_configured_ones(tmp_path) -> N AptModule().apply(context) - disabled_top_level = (dst / "etc/apt/sources.list").read_text(encoding="utf-8") + # No configured source declares a uri_to_replace, so none of the existing + # entries are disabled: they are copied through unchanged. + top_level = (dst / "etc/apt/sources.list").read_text(encoding="utf-8") assert ( - disabled_top_level - == "# already disabled\n# deb http://example.invalid/legacy stable main\n\n" + top_level + == "# already disabled\ndeb http://example.invalid/legacy stable main\n\n" ) - disabled_list = (dst / "etc/apt/sources.list.d/extra.list").read_text( - encoding="utf-8" - ) - assert disabled_list == "# deb http://example.invalid/extra stable main\n" + extra_list = (dst / "etc/apt/sources.list.d/extra.list").read_text(encoding="utf-8") + assert extra_list == "deb http://example.invalid/extra stable main\n" - disabled_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text( + docker_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text( encoding="utf-8" ) - assert disabled_sources == ( - "Types: deb\nURIs: http://example.invalid/docker\nSuites: stable\nEnabled: no" + assert docker_sources == ( + "Types: deb\nURIs: http://example.invalid/docker\nSuites: stable\nEnabled: yes" "\n\n" - "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\nEnabled: no\n" + "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n" ) assert not (dst / "etc/apt/sources.list.d/ignored.txt").exists() @@ -133,6 +135,55 @@ def test_apply_disables_existing_sources_and_adds_configured_ones(tmp_path) -> N ) +def test_apply_disables_only_existing_sources_matching_uri_to_replace( + tmp_path, +) -> None: + src = tmp_path / "src" + apt_dir = src / "etc/apt" + sources_list_d = apt_dir / "sources.list.d" + sources_list_d.mkdir(parents=True) + + (apt_dir / "sources.list").write_text( + "deb http://example.invalid/legacy stable main\n", encoding="utf-8" + ) + (sources_list_d / "extra.list").write_text( + "deb http://example.invalid/extra stable main\n", encoding="utf-8" + ) + (sources_list_d / "docker.sources").write_text( + "Types: deb\nURIs: http://example.invalid/extra\nSuites: stable\nEnabled: yes\n" + "\n" + "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n" + "\n\n", + encoding="utf-8", + ) + + dst = tmp_path / "dst" + context = Context( + config=make_config_with({"uri-to-replace": "http://example.invalid/extra"}), + dst=dst, + src=src, + ) + + AptModule().apply(context) + + # The legacy source doesn't match the pattern, so it stays enabled... + top_level = (dst / "etc/apt/sources.list").read_text(encoding="utf-8") + assert top_level == "deb http://example.invalid/legacy stable main\n" + + # ...but every entry whose URI matches is disabled, regardless of format. + extra_list = (dst / "etc/apt/sources.list.d/extra.list").read_text(encoding="utf-8") + assert extra_list == "# deb http://example.invalid/extra stable main\n" + + docker_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text( + encoding="utf-8" + ) + assert docker_sources == ( + "Types: deb\nURIs: http://example.invalid/extra\nSuites: stable\nEnabled: no" + "\n\n" + "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n" + ) + + def test_apply_without_existing_top_level_sources_list(tmp_path) -> None: src = tmp_path / "src" (src / "etc/apt").mkdir(parents=True) @@ -200,6 +251,106 @@ def test_apply_without_signed_by_writes_no_field(tmp_path) -> None: ) +def test_apply_skips_source_when_uri_to_replace_matches_nothing(tmp_path) -> None: + src = tmp_path / "src" + apt_dir = src / "etc/apt" + apt_dir.mkdir(parents=True) + (apt_dir / "sources.list").write_text( + "deb http://example.invalid/legacy stable main\n", encoding="utf-8" + ) + + dst = tmp_path / "dst" + context = Context( + config=make_config_with({"uri-to-replace": "http://other.invalid/*"}), + dst=dst, + src=src, + ) + + AptModule().apply(context) + + assert not (dst / "etc/apt/sources.list.d/interdiode.sources").exists() + + +def test_apply_writes_source_when_uri_to_replace_matches_existing_uri( + tmp_path, +) -> None: + src = tmp_path / "src" + apt_dir = src / "etc/apt" + apt_dir.mkdir(parents=True) + (apt_dir / "sources.list").write_text( + "deb http://legacy.invalid/debian stable main\n", encoding="utf-8" + ) + + dst = tmp_path / "dst" + context = Context( + config=make_config_with({"uri-to-replace": "http://legacy.invalid/*"}), + dst=dst, + src=src, + ) + + AptModule().apply(context) + + assert (dst / "etc/apt/sources.list.d/interdiode.sources").exists() + + +def test_apply_skips_source_when_uri_to_replace_set_without_src(tmp_path) -> None: + dst = tmp_path / "dst" + context = Context( + config=make_config_with({"uri-to-replace": "http://*"}), dst=dst, src=None + ) + + AptModule().apply(context) + + assert not (dst / "etc/apt/sources.list.d/interdiode.sources").exists() + + +def test_apply_matches_uri_to_replace_across_list_line_variants(tmp_path) -> None: + src = tmp_path / "src" + apt_dir = src / "etc/apt" + sources_list_d = apt_dir / "sources.list.d" + sources_list_d.mkdir(parents=True) + + (apt_dir / "sources.list").write_text( + "# a comment\n" + "\n" + "garbage line here\n" + "deb [unterminated bracket with no closing marker\n" + "deb [arch=amd64 trusted=yes] http://example.invalid/extra stable main\n", + encoding="utf-8", + ) + (sources_list_d / "ignored.txt").write_text("not a source file\n", encoding="utf-8") + + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "apt": { + "sources": [ + { + "name": "first", + "uris": ["http://example.invalid/first"], + "suites": ["stable"], + "uri-to-replace": "http://example.invalid/extra", + }, + { + "name": "second", + "uris": ["http://example.invalid/second"], + "suites": ["stable"], + "uri-to-replace": "http://example.invalid/extra", + }, + ], + } + }, + } + dst = tmp_path / "dst" + context = Context(config=Configuration.model_validate(data), dst=dst, src=src) + + AptModule().apply(context) + + assert (dst / "etc/apt/sources.list.d/first.sources").exists() + assert (dst / "etc/apt/sources.list.d/second.sources").exists() + + def test_apply_sanitizes_source_name_with_slash(tmp_path) -> None: dst = tmp_path / "dst" context = Context( diff --git a/tests/test_yum.py b/tests/test_yum.py index fa727a4..dd10c12 100644 --- a/tests/test_yum.py +++ b/tests/test_yum.py @@ -126,7 +126,9 @@ def test_apply_unsigned_repo_disables_gpgcheck(tmp_path) -> None: ) -def test_apply_disables_existing_repos_before_adding_configured_ones(tmp_path) -> None: +def test_apply_copies_existing_repos_unchanged_without_index_url_to_replace( + tmp_path, +) -> None: src = tmp_path / "src" repos_dir = src / "etc/yum.repos.d" repos_dir.mkdir(parents=True) @@ -148,13 +150,13 @@ def test_apply_disables_existing_repos_before_adding_configured_ones(tmp_path) - YumModule().apply(context) - disabled_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8") - assert "[old-one]" in disabled_multi - assert "[old-two]" in disabled_multi + # No configured repository declares an index_url_to_replace, so the + # existing sections are copied through unchanged, still enabled. + copied_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8") parser = ConfigParser(interpolation=None) - parser.read_string(disabled_multi) - assert parser["old-one"]["enabled"] == "0" - assert parser["old-two"]["enabled"] == "0" + parser.read_string(copied_multi) + assert parser["old-one"]["enabled"] == "1" + assert parser["old-two"]["enabled"] == "1" assert not (dst / "etc/yum.repos.d/ignored.txt").exists() @@ -165,6 +167,152 @@ def test_apply_disables_existing_repos_before_adding_configured_ones(tmp_path) - ) +def test_apply_disables_only_existing_sections_matching_index_url_to_replace( + tmp_path, +) -> None: + src = tmp_path / "src" + repos_dir = src / "etc/yum.repos.d" + repos_dir.mkdir(parents=True) + (repos_dir / "multi.repo").write_text( + "[old-one]\nname=old-one\nbaseurl=https://old-one.invalid\nenabled=1\ngpgcheck=1\n" + "\n" + "[old-two]\nname=old-two\nbaseurl=https://old-two.invalid\nenabled=1\n", + encoding="utf-8", + ) + + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "yum": { + "repositories": [ + { + "name": "unsigned", + "index-url": "https://example.invalid/yum", + "index-url-to-replace": "https://old-one.invalid", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=make_config(data), dst=dst, src=src) + + YumModule().apply(context) + + disabled_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8") + parser = ConfigParser(interpolation=None) + parser.read_string(disabled_multi) + assert parser["old-one"]["enabled"] == "0" + assert parser["old-two"]["enabled"] == "1" + + +def test_apply_skips_repository_when_index_url_to_replace_matches_nothing( + tmp_path, +) -> None: + src = tmp_path / "src" + repos_dir = src / "etc/yum.repos.d" + repos_dir.mkdir(parents=True) + (repos_dir / "old.repo").write_text( + "[old]\nname=old\nbaseurl=https://old.invalid\nenabled=1\n", encoding="utf-8" + ) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "yum": { + "repositories": [ + { + "name": "unsigned", + "index-url": "https://example.invalid/yum", + "index-url-to-replace": "https://other.invalid/*", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=make_config(data), dst=dst, src=src) + + YumModule().apply(context) + + assert not (dst / "etc/yum.repos.d/unsigned.repo").exists() + + +def test_apply_writes_repository_when_index_url_to_replace_matches_existing( + tmp_path, +) -> None: + src = tmp_path / "src" + repos_dir = src / "etc/yum.repos.d" + repos_dir.mkdir(parents=True) + (repos_dir / "old.repo").write_text( + "[old]\nname=old\nbaseurl=https://old.invalid/yum\nenabled=1\n", + encoding="utf-8", + ) + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "yum": { + "repositories": [ + { + "name": "unsigned", + "index-url": "https://example.invalid/yum", + "index-url-to-replace": "https://old.invalid/*", + } + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=make_config(data), dst=dst, src=src) + + YumModule().apply(context) + + assert (dst / "etc/yum.repos.d/unsigned.repo").exists() + + +def test_apply_matches_index_url_to_replace_with_unrelated_files_and_multiple_repos( + tmp_path, +) -> None: + src = tmp_path / "src" + repos_dir = src / "etc/yum.repos.d" + repos_dir.mkdir(parents=True) + (repos_dir / "old.repo").write_text( + "[old]\nname=old\nbaseurl=https://old.invalid/yum\nenabled=1\n", + encoding="utf-8", + ) + (repos_dir / "ignored.txt").write_text("not a repo file\n", encoding="utf-8") + + data = { + "uuid": "12345678-1234-5678-1234-567812345678", + "organisation": "InterDiode", + "configurations": { + "yum": { + "repositories": [ + { + "name": "first", + "index-url": "https://example.invalid/first", + "index-url-to-replace": "https://old.invalid/*", + }, + { + "name": "second", + "index-url": "https://example.invalid/second", + "index-url-to-replace": "https://old.invalid/*", + }, + ] + } + }, + } + dst = tmp_path / "dst" + context = Context(config=make_config(data), dst=dst, src=src) + + YumModule().apply(context) + + assert (dst / "etc/yum.repos.d/first.repo").exists() + assert (dst / "etc/yum.repos.d/second.repo").exists() + + def test_apply_sanitizes_repository_and_key_names_with_slash( tmp_path, monkeypatch ) -> None: