diff --git a/CHANGELOG.md b/CHANGELOG.md index e590d85..8b0b69a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ recorded here; their history is the git log. ## [Unreleased] +## [4.22.0] - 2026-10-23 + +### Added +- `AptProxyMixin` (`napalm_device_types.apt_proxy`): apt's proxy on a + Debian-family host, for a driver to mix in (NetOrk/netork#1397). + - `get_package_proxy()` is the proxy apt really fetches through, as + `apt-config dump` says, and no longer the content of one file. A proxy the + Debian or Ubuntu installer set elsewhere wins over `00proxy`, and was missed. + - `get_package_proxy_report()` adds where it is set: every line of apt's + configuration that sets it, in the order apt reads them, the per-host + proxies, and the sources fetched over http and over https. + - `check_package_proxy(url, expect)` fetches a page from the host with apt's + own downloader, past any proxy, and says whether the host reached it and + whether the page says what you expected. + - `configure_package_proxy(url)` writes the proxy, comments out every other + line that sets a different one, reads the result back and fetches one + source through the proxy. It fails when apt still fetches elsewhere. + - `same_proxy_url()` compares two proxy URLs by scheme, host and port. + ## [4.21.1] - 2026-10-22 The first release on PyPI since 0.1.0. diff --git a/README.md b/README.md index b675bbf..ce16fd1 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,15 @@ that binary links against. An application needs it to let a statement about a se (the postgresql unit) reach the findings on its packages. A driver supplies `_run_service_packages_command`. +`AptProxyMixin` (`napalm_device_types.apt_proxy`, since 4.22) is `PackageProxyMixin` for a +Debian-family host, mixed in the same way. What apt fetches through is what `apt-config dump` +says, not one file: apt reads `apt.conf.d` in order and `apt.conf` last, and the last value +wins (`get_package_proxy`, `get_package_proxy_report` with every line that sets it). +`check_package_proxy(url, expect)` fetches a page with apt's own downloader, past any proxy. +`configure_package_proxy(url)` writes `00proxy`, comments out every other line that sets a +different proxy, reads the result back and fetches one http source's `InRelease` through it. +A driver supplies `_run_apt_command`, `_run_apt_root_command` and `_write_apt_file`. + `HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer kernel of the running flavour installed), whether it patches itself (unattended-upgrades, diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 5dbda76..f0d0274 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -65,6 +65,13 @@ Introspection -- :func:`~napalm_device_types.roles.roles_of`, """ from napalm_device_types.access_point import AccessPointDriver +from napalm_device_types.apt_proxy import ( + APT_PROXY_COMMAND, + AptProxyMixin, + parse_apt_fetch, + parse_apt_proxy, + same_proxy_url, +) from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec, rules_max_weight from napalm_device_types.channel import ( ByteStream, @@ -225,6 +232,11 @@ __all__ = [ "FixStepDict", "FixStepOutcomeDict", "PackageProxyMixin", + "APT_PROXY_COMMAND", + "AptProxyMixin", + "parse_apt_fetch", + "parse_apt_proxy", + "same_proxy_url", "SnmpAgentMixin", "SnmpAgentSettingsDict", "run_fix_steps", diff --git a/napalm_device_types/apt_proxy.py b/napalm_device_types/apt_proxy.py new file mode 100644 index 0000000..7cc7b6f --- /dev/null +++ b/napalm_device_types/apt_proxy.py @@ -0,0 +1,458 @@ +"""apt's proxy: what apt fetches through, whether a proxy answers, and pointing apt at one. + +Read the same way on every Debian-family host, so the commands and their parse +live here once, like :mod:`~napalm_device_types.host_status`; a driver only +carries a command across and runs one as root. + +**What apt fetches through** is what ``apt-config dump`` says, not one file. apt +reads every file in ``Dir::Etc::parts`` (``apt.conf.d``) in byte order, then +``Dir::Etc::main`` (``apt.conf``); for a single value the last one read wins. The +Debian installer writes its proxy into ``apt.conf``, Ubuntu's into +``apt.conf.d/90curtin-aptproxy``: both win over an ``00proxy`` +(NetOrk/netork#1397). ``DIRECT`` for the proxy, or for one host +(``Acquire::http::Proxy::``), means no proxy there. + +**Where it is set** is every line of apt's files that sets the http proxy, in the +order apt reads them. A line of the form ``Acquire::http::Proxy "…";`` can be +commented out; a setting inside a block (``Acquire::http { Proxy "…"; };``) is +named but left alone. apt skips a file in ``apt.conf.d`` whose name has other +characters than letters, digits, ``_``, ``-`` and ``.``, or an extension other +than ``.conf`` (``00proxy.bak``, ``*.dpkg-old``); so is it here. + +**Whether a proxy answers** is a fetch with apt's own downloader, +``apt-helper download-file``, which is on every apt host and needs no root: +straight to a page the caller names (past any proxy), or through the proxy apt is +set to. An HTTP error is an answer too: the other end was reached. Whether the +page is the right one, the caller says (``expect``). + +**Pointing apt at a proxy** writes :attr:`AptProxyMixin.APT_PROXY_FILE`, comments +out every other line that sets a different one, reads the result back from +``apt-config``, and fetches the ``InRelease`` of one http source through it. The +last step is what proves it: apt reached a mirror through the proxy it is now set +to. The source chosen is one apt does not fetch ``DIRECT``. +""" + +from __future__ import annotations + +import re +import shlex +from typing import TYPE_CHECKING +from urllib.parse import urlsplit + +from typing_extensions import TypedDict + +from napalm_device_types.fixes import PackageProxyMixin, run_fix_steps +from napalm_device_types.models import FixResultDict, FixStepOutcomeDict +from napalm_device_types.terminal import strip_terminal_codes + + +class AptProxySettingDict(TypedDict): + """One line of apt's configuration that sets the http proxy (``AptProxyMixin``).""" + + file: str + line: int # 1-based + value: str + #: True for ``Acquire::http::Proxy "…";`` on a line of its own, which can be + #: commented out; False for a setting inside a block. + editable: bool + + +class AptProxyReportDict(TypedDict): + """What apt fetches through, and where that is set (``get_package_proxy_report``).""" + + #: The http proxy apt fetches through; None for none or ``DIRECT``. + proxy: str | None + #: ``Acquire::http::Proxy::``: a proxy, or ``DIRECT``, for one host. + per_host: dict[str, str] + #: ``Acquire::http::Proxy-Auto-Detect``: a program that names the proxy. + auto_detect: str | None + #: Every line that sets it, in the order apt reads them; the last one wins. + settings: list[AptProxySettingDict] + #: The base URIs of the host's sources over http and over https. + http_sources: list[str] + https_sources: list[str] + + +class ProxyFetchDict(TypedDict): + """A fetch from the host with apt's downloader (``check_package_proxy``).""" + + #: Whether the other end answered at all; an HTTP error is an answer. + reachable: bool + #: Whether the page contained what the caller expected; None when it named nothing + #: or nothing came back. + expected: bool | None + #: apt's reason when the fetch failed, else "". + error: str + + +_BEGIN = "APROXY_BEGIN" +_END = "APROXY_END" +_APT_HELPER = "/usr/lib/apt/apt-helper" + +#: A proxy URL apt's file and a shell word can both carry. +PROXY_URL = re.compile(r"^https?://[A-Za-z0-9.\-\[\]:@%_~/]+$") + +#: One line, POSIX ``sh``, read-only, no privileges. In a subshell, so neither the +#: locale nor its variables stay behind in an interactive session; the frame +#: markers are printed in two halves so that an echoing transport does not show +#: them early. +APT_PROXY_COMMAND = ( + "( export LC_ALL=C; printf '%s%s\\n' APROXY_ BEGIN; " + "echo '[dump]'; apt-config dump 2>/dev/null | grep -i '^Acquire::http::Proxy'; " + 'eval "$(apt-config shell ap_parts Dir::Etc::parts/d ap_main Dir::Etc::main/f 2>/dev/null)"; ' + "echo '[parts]'; for f in \"${ap_parts:-/etc/apt/apt.conf.d/}\"*; do " + '[ -f "$f" ] && grep -Hin proxy "$f"; done; ' + 'echo \'[main]\'; f="${ap_main:-/etc/apt/apt.conf}"; [ -f "$f" ] && grep -Hin proxy "$f"; ' + "echo '[sources]'; apt-get indextargets --no-release-info --format '$(BASE_URI)' " + "2>/dev/null | sort -u; " + "printf '%s%s\\n' APROXY_ END ) 2>/dev/null | cat" +) + +_DUMP = re.compile(r'^Acquire::http::Proxy(::\S+|-Auto-Detect)?\s+"([^"]*)"', re.I) +_GREP = re.compile(r"^(/[^:]+):(\d+):(.*)$") +_FLAT = re.compile(r'^\s*Acquire::http::Proxy\s+"([^"]*)"\s*;', re.I) +_NOT_HTTP_PROXY = re.compile( + r"^\s*(//|#)|Acquire::(https|ftp|socks)\b|::Proxy::|Proxy-Auto-Detect", re.I +) +_IN_BLOCK = re.compile(r'\bProxy\s+"([^"]*)"', re.I) +#: The names apt reads in ``apt.conf.d``: no extension, or ``.conf``. +_PART_NAME = re.compile(r"^(?:[A-Za-z0-9_\-]+|[A-Za-z0-9_.\-]+\.conf)$") +_NO_PROXY = frozenset({"", "direct"}) + + +def _sections(output: str) -> dict[str, list[str]]: + lines = [line.rstrip() for line in strip_terminal_codes(output).splitlines()] + stripped = [line.strip() for line in lines] + try: + start = stripped.index(_BEGIN) + end = stripped.index(_END, start) + except ValueError: + raise ValueError("no intact apt proxy report in the output") from None + sections: dict[str, list[str]] = {} + current: list[str] = [] + for line, bare in zip(lines[start + 1 : end], stripped[start + 1 : end], strict=True): + if bare.startswith("[") and bare.endswith("]") and " " not in bare: + current = sections.setdefault(bare[1:-1], []) + elif bare: + current.append(line) + return sections + + +def _setting(line: str, main: bool) -> AptProxySettingDict | None: + found = _GREP.match(line) + if not found: + return None + path, number, text = found.groups() + if not main and not _PART_NAME.match(path.rsplit("/", 1)[-1]): + return None + flat = _FLAT.match(text) + if flat: + return {"file": path, "line": int(number), "value": flat.group(1), "editable": True} + if _NOT_HTTP_PROXY.search(text): + return None + nested = _IN_BLOCK.search(text) + if not nested: + return None + return {"file": path, "line": int(number), "value": nested.group(1), "editable": False} + + +def _dump(lines: list[str]) -> tuple[str | None, dict[str, str], str | None]: + """The proxy, the per-host proxies and the auto-detect program in ``apt-config dump``.""" + proxy: str | None = None + per_host: dict[str, str] = {} + auto_detect: str | None = None + for found in (_DUMP.match(line.strip()) for line in lines): + if not found: + continue + suffix, value = found.groups() + if not suffix: + proxy = None if value.lower() in _NO_PROXY else value + elif suffix.lower() == "-auto-detect": + auto_detect = value or None + else: + per_host[suffix[2:]] = value + return proxy, per_host, auto_detect + + +def parse_apt_proxy(output: str) -> AptProxyReportDict: + """The report :data:`APT_PROXY_COMMAND` prints, as :class:`AptProxyReportDict`. + + :raises ValueError: if the output carried no intact report. + """ + sections = _sections(output) + proxy, per_host, auto_detect = _dump(sections.get("dump", [])) + settings = [ + s + for main, key in ((False, "parts"), (True, "main")) + for s in (_setting(line, main) for line in sections.get(key, [])) + if s + ] + sources = [line.strip() for line in sections.get("sources", [])] + return { + "proxy": proxy, + "per_host": per_host, + "auto_detect": auto_detect, + "settings": settings, + "http_sources": [s for s in sources if s.startswith("http://")], + "https_sources": [s for s in sources if s.startswith("https://")], + } + + +_DEFAULT_PORT = {"http": 80, "https": 443} + + +def _endpoint(url: str | None) -> tuple[str, str, int | None, str] | None: + """Scheme, host, port (the scheme's own when none is given) and path; None for no URL.""" + if not url: + return None + try: + parts = urlsplit(url) + port = parts.port or _DEFAULT_PORT.get(parts.scheme) + except ValueError: + return None + return parts.scheme, parts.hostname or "", port, parts.path.rstrip("/") + + +def same_proxy_url(a: str | None, b: str | None) -> bool: + """Whether *a* and *b* name the same proxy: scheme, host and port, a trailing ``/`` aside.""" + first = _endpoint(a) + return first is not None and first == _endpoint(b) + + +def apt_fetch_command(url: str, *, direct: bool = False, expect: str = "") -> str: + """The command that fetches *url* with apt's downloader into a temporary file. + + *direct* goes past any proxy apt is set to, to *url*'s host itself; without it + the fetch takes apt's own way. *expect* is text the page should contain. + + :raises ValueError: if *url* is not an http URL a shell word can carry. + """ + if not PROXY_URL.match(url) or not url.startswith("http://"): + raise ValueError(f"not an http URL: {url!r}") + options = ["-o Acquire::Retries=0", "-o Acquire::http::Timeout=10"] + host = urlsplit(url).hostname or "" + if direct and host and ":" not in host: + # *host* comes from a URL PROXY_URL let through: no quote in it. + options.append(f"-o 'Acquire::http::Proxy::{host}=DIRECT'") + match = f"echo '[match]'; grep -c -F -- {shlex.quote(expect)} \"$t\"; " if expect else "" + return ( + "( export LC_ALL=C; t=$(mktemp) || exit 1; printf '%s%s\\n' APROXY_ BEGIN; " + f"echo '[fetch]'; {_APT_HELPER} {' '.join(options)} download-file " + f'{shlex.quote(url)} "$t" 2>&1; rc=$?; echo \'[rc]\'; echo "$rc"; ' + f"{match}rm -f \"$t\"; printf '%s%s\\n' APROXY_ END ) | cat" + ) + + +_FAILED = re.compile(r"^E: Failed to fetch \S+\s+(.*)$") +_HTTP_STATUS = re.compile(r"^\d{3}\s") + + +def parse_apt_fetch(output: str) -> ProxyFetchDict: + """What :func:`apt_fetch_command` printed, as :class:`ProxyFetchDict`. + + :raises ValueError: if the output carried no intact report, or apt-helper is + missing -- then nothing could be tried. + """ + sections = _sections(output) + try: + rc = int(sections["rc"][0]) + except (KeyError, IndexError, ValueError): + raise ValueError("no exit status in the apt fetch report") from None + if rc in (126, 127): + raise ValueError(f"{_APT_HELPER} cannot run on this host") + match = sections.get("match") + if rc == 0: + expected = (match[0].strip() not in ("", "0")) if match else None + return {"reachable": True, "expected": expected, "error": ""} + error = next( + (m.group(1).strip() for m in map(_FAILED.match, sections.get("fetch", [])) if m), + "the download failed", + ) + answered = bool(_HTTP_STATUS.match(error)) + return {"reachable": answered, "expected": False if answered else None, "error": error} + + +#: sed's edit of one line: comment it out, if it still sets the http proxy. +_COMMENT_OUT = "-e '{line}{{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}}'" + + +def comment_out_command(path: str, lines: list[int]) -> str: + """The root command that comments out *lines* of *path* where they set the http proxy.""" + edits = " ".join(_COMMENT_OUT.format(line=int(n)) for n in lines) + return f"sed -i {edits} {shlex.quote(path)}" + + +def _editable_lines(settings: list[AptProxySettingDict]) -> dict[str, list[int]]: + by_file: dict[str, list[int]] = {} + for s in settings: + if s["editable"]: + by_file.setdefault(s["file"], []).append(s["line"]) + return by_file + + +def _described(s: AptProxySettingDict) -> str: + what = "commented out" if s["editable"] else "left alone (inside a block)" + return f"{s['file']}:{s['line']} ({s['value']}) {what}" + + +class AptProxyMixin(PackageProxyMixin): + """:class:`~napalm_device_types.fixes.PackageProxyMixin` for a Debian-family host. + + The template form, like :class:`~napalm_device_types.host_status.HostStatusMixin`: + the reading, the parse and the order of the fix are the same everywhere; a + driver supplies the transports below. Mixed in by the drivers that can, so + ``callable(getattr(driver, "configure_package_proxy", None))`` stays a truthful + answer. + """ + + #: The file the proxy is written to. Its name carries no application's. + APT_PROXY_FILE = "/etc/apt/apt.conf.d/00proxy" + + if TYPE_CHECKING: # pragma: no cover - declared for type checkers only + + def _run_apt_command(self, command: str) -> str: + """Run *command* on the host with ``sh``, without privileges; what it printed.""" + ... + + def _run_apt_root_command(self, command: str, done: str) -> FixStepOutcomeDict: + """Run *command* as root; ``ok`` with *done* when it exited 0.""" + ... + + def _write_apt_file(self, path: str, content: str) -> FixStepOutcomeDict: + """Replace *path* with *content*, as a new file owned by root, mode 644.""" + ... + + def _is_apt_host(self) -> bool: + """Whether apt is the host's package manager; a driver that can tell overrides it.""" + return True + + def get_package_proxy_report(self) -> AptProxyReportDict | None: + """What apt fetches through and where that is set; None on a host without apt. + + * proxy (string or None): the http proxy apt fetches through + * per_host ({host: proxy or "DIRECT"}) + * auto_detect (string or None): ``Acquire::http::Proxy-Auto-Detect`` + * settings (list of {file, line, value, editable}), in apt's reading order + * http_sources, https_sources (lists of the sources' base URIs) + + :raises ValueError: if the host's output carried no intact report. + """ + if not self._is_apt_host(): + return None + return parse_apt_proxy(self._run_apt_command(APT_PROXY_COMMAND)) + + def get_package_proxy(self) -> str | None: + """The proxy apt fetches through now; None for none, or a host without apt.""" + report = self.get_package_proxy_report() + return report["proxy"] if report else None + + def check_package_proxy(self, url: str, expect: str = "") -> ProxyFetchDict: + """Fetch *url* from the host, past any proxy apt is set to. + + Whether the host reaches it, and whether the page contains *expect*. + + :raises ValueError: if *url* is no http URL, or the host cannot fetch at all. + :raises NotImplementedError: on a host without apt. + """ + if not self._is_apt_host(): + raise NotImplementedError("not an apt host") + return parse_apt_fetch( + self._run_apt_command(apt_fetch_command(url, direct=True, expect=expect)) + ) + + def configure_package_proxy(self, url: str) -> FixResultDict: + """apt fetches through *url*, and is shown to; ``""`` removes :attr:`APT_PROXY_FILE`.""" + if not url: + return run_fix_steps([("configure", self._remove_package_proxy)]) + return run_fix_steps( + [ + ("configure", lambda: self._write_package_proxy(url)), + ("conflicts", lambda: self._comment_out_other_proxies(url)), + ("verify", lambda: self._verify_package_proxy(url)), + ("fetch", lambda: self._fetch_through_package_proxy(url)), + ] + ) + + def _not_apt(self) -> FixStepOutcomeDict: + return {"ok": False, "output": "not an apt host"} + + def _remove_package_proxy(self) -> FixStepOutcomeDict: + if not self._is_apt_host(): + return self._not_apt() + return self._run_apt_root_command( + f"rm -f {shlex.quote(self.APT_PROXY_FILE)}", "proxy removed" + ) + + def _write_package_proxy(self, url: str) -> FixStepOutcomeDict: + if not self._is_apt_host(): + return self._not_apt() + if not PROXY_URL.match(url): + raise ValueError(f"not a proxy URL: {url!r}") + return self._write_apt_file(self.APT_PROXY_FILE, f'Acquire::http::Proxy "{url}";\n') + + def _report(self) -> AptProxyReportDict: + # The steps after "configure" run only once it passed, on an apt host. + return parse_apt_proxy(self._run_apt_command(APT_PROXY_COMMAND)) + + def _comment_out_other_proxies(self, url: str) -> FixStepOutcomeDict: + others = [ + s + for s in self._report()["settings"] + if s["file"] != self.APT_PROXY_FILE and not same_proxy_url(s["value"], url) + ] + for path, numbers in _editable_lines(others).items(): + outcome = self._run_apt_root_command( + comment_out_command(path, numbers), f"commented out in {path}" + ) + if not outcome["ok"]: + return outcome + return {"ok": True, "output": "; ".join(map(_described, others)) or "no other proxy set"} + + def _verify_package_proxy(self, url: str) -> FixStepOutcomeDict: + report = self._report() + if same_proxy_url(report["proxy"], url): + return {"ok": True, "output": f"apt fetches through {url}"} + where = [ + f"{s['file']}:{s['line']}" + for s in report["settings"] + if s["value"] == report["proxy"] and s["file"] != self.APT_PROXY_FILE + ] + return { + "ok": False, + "output": f"apt still fetches through {report['proxy'] or 'no proxy'}" + + (f", set in {', '.join(where)}" if where else ""), + } + + def _fetch_through_package_proxy(self, url: str) -> FixStepOutcomeDict: + report = self._report() + direct = {h.lower() for h, v in report["per_host"].items() if v.lower() == "direct"} + source = next( + ( + s + for s in report["http_sources"] + if "/dists/" in s and (urlsplit(s).hostname or "").lower() not in direct + ), + None, + ) + if source is None: + return {"ok": True, "output": "no http source to fetch through the proxy; not tried"} + target = source.rstrip("/") + "/InRelease" + fetched = parse_apt_fetch(self._run_apt_command(apt_fetch_command(target))) + if fetched["reachable"] and not fetched["error"]: + return {"ok": True, "output": f"fetched {target} through {url}"} + return {"ok": False, "output": f"{target}: {fetched['error']}"} + + +__all__ = [ + "APT_PROXY_COMMAND", + "AptProxyMixin", + "AptProxyReportDict", + "AptProxySettingDict", + "ProxyFetchDict", + "PROXY_URL", + "apt_fetch_command", + "comment_out_command", + "parse_apt_fetch", + "parse_apt_proxy", + "same_proxy_url", +] diff --git a/pyproject.toml b/pyproject.toml index 6d850cb..66b9c88 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "4.21.1" +version = "4.22.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_apt_proxy.py b/tests/test_apt_proxy.py new file mode 100644 index 0000000..3071f2d --- /dev/null +++ b/tests/test_apt_proxy.py @@ -0,0 +1,533 @@ +"""apt's proxy: what apt fetches through, whether a proxy answers, pointing apt at one. + +apt reads every file in ``apt.conf.d`` in order and ``apt.conf`` last; the last +value wins. The Debian installer writes its proxy into ``apt.conf``, Ubuntu's +into ``apt.conf.d/90curtin-aptproxy``: reading one file was not enough +(NetOrk/netork#1397). The fetch outputs are apt-helper's own, from a Mint 22.3 +host against netOrk's test server. +""" + +from __future__ import annotations + +import os +import re +import subprocess + +import pytest + +from napalm_device_types.apt_proxy import ( + APT_PROXY_COMMAND, + AptProxyMixin, + apt_fetch_command, + comment_out_command, + parse_apt_fetch, + parse_apt_proxy, + same_proxy_url, +) +from napalm_device_types.fixes import PackageProxyMixin + + +def _report(dump=(), parts=(), main=(), sources=()): + lines = ["APROXY_BEGIN", "[dump]", *dump, "[parts]", *parts, "[main]", *main] + return "\n".join([*lines, "[sources]", *sources, "APROXY_END"]) + + +_D = "/etc/apt/apt.conf.d/" + + +class TestEffectiveProxy: + def test_the_proxy_apt_fetches_through(self): + report = parse_apt_proxy( + _report( + dump=['Acquire::http::Proxy "http://c:3142";'], + parts=[f'{_D}00proxy:1:Acquire::http::Proxy "http://c:3142";'], + ) + ) + + assert report["proxy"] == "http://c:3142" + assert report["settings"] == [ + {"file": f"{_D}00proxy", "line": 1, "value": "http://c:3142", "editable": True} + ] + + def test_none_without_a_proxy(self): + report = parse_apt_proxy(_report()) + + assert report["proxy"] is None + assert report["settings"] == [] + assert report["per_host"] == {} + assert report["auto_detect"] is None + + @pytest.mark.parametrize("value", ["", "DIRECT", "direct"]) + def test_an_empty_or_direct_proxy_is_none(self, value): + assert parse_apt_proxy(_report(dump=[f'Acquire::http::Proxy "{value}";']))["proxy"] is None + + def test_keys_are_case_insensitive(self): + report = parse_apt_proxy(_report(dump=['acquire::HTTP::proxy "http://c:3142";'])) + assert report["proxy"] == "http://c:3142" + + def test_per_host_overrides_and_auto_detect(self): + report = parse_apt_proxy( + _report( + dump=[ + 'Acquire::http::Proxy "http://c:3142";', + 'Acquire::http::Proxy::deb.debian.org "DIRECT";', + 'Acquire::http::Proxy::ppa.launchpad.net "http://other:3128";', + 'Acquire::http::Proxy-Auto-Detect "/usr/bin/auto-apt-proxy";', + ] + ) + ) + + assert report["proxy"] == "http://c:3142" + assert report["per_host"] == { + "deb.debian.org": "DIRECT", + "ppa.launchpad.net": "http://other:3128", + } + assert report["auto_detect"] == "/usr/bin/auto-apt-proxy" + + def test_no_intact_report_raises(self): + with pytest.raises(ValueError): + parse_apt_proxy("APROXY_BEGIN\n[dump]\n") + + +class TestWhereItIsSet: + def test_every_setting_in_apt_s_reading_order(self): + report = parse_apt_proxy( + _report( + dump=['Acquire::http::Proxy "http://old:3128";'], + parts=[ + f'{_D}00proxy:1:Acquire::http::Proxy "http://c:3142";', + f'{_D}90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";', + ], + main=['/etc/apt/apt.conf:4: Acquire::http::Proxy "http://older:8080";'], + ) + ) + + assert [(s["file"], s["line"], s["value"]) for s in report["settings"]] == [ + (f"{_D}00proxy", 1, "http://c:3142"), + (f"{_D}90curtin-aptproxy", 1, "http://old:3128"), + ("/etc/apt/apt.conf", 4, "http://older:8080"), + ] + + def test_a_nested_setting_is_named_but_not_editable(self): + report = parse_apt_proxy(_report(parts=[f'{_D}01proxy:2: Proxy "http://nested:3142";'])) + + assert report["settings"] == [ + {"file": f"{_D}01proxy", "line": 2, "value": "http://nested:3142", "editable": False} + ] + + @pytest.mark.parametrize( + "line", + [ + '// Acquire::http::Proxy "http://c:3142";', + '# Acquire::http::Proxy "http://c:3142";', + 'Acquire::https::Proxy "http://c:3142";', + 'Acquire::ftp::Proxy "http://c:3142";', + 'Acquire::http::Proxy::deb.debian.org "DIRECT";', + 'Acquire::http::Proxy-Auto-Detect "/usr/bin/x";', + "Acquire::http {", + ], + ) + def test_what_does_not_set_the_http_proxy_is_left_out(self, line): + assert parse_apt_proxy(_report(parts=[f"{_D}00x:1:{line}"]))["settings"] == [] + + @pytest.mark.parametrize("name", ["00proxy.bak", "00proxy.dpkg-old", "00proxy.save", "a b"]) + def test_files_apt_does_not_read_are_left_out(self, name): + line = f'{_D}{name}:1:Acquire::http::Proxy "http://c:3142";' + assert parse_apt_proxy(_report(parts=[line]))["settings"] == [] + + def test_a_conf_extension_is_read(self): + line = f'{_D}01proxy.conf:1:Acquire::http::Proxy "http://c:3142";' + assert len(parse_apt_proxy(_report(parts=[line]))["settings"]) == 1 + + +class TestSources: + def test_https_sources_bypass_an_http_proxy(self): + report = parse_apt_proxy( + _report( + sources=[ + "http://archive.ubuntu.com/ubuntu/dists/noble/", + "https://download.docker.com/linux/ubuntu/dists/noble/", + ] + ) + ) + + assert report["http_sources"] == ["http://archive.ubuntu.com/ubuntu/dists/noble/"] + assert report["https_sources"] == ["https://download.docker.com/linux/ubuntu/dists/noble/"] + + +class TestSameProxyUrl: + @pytest.mark.parametrize( + "a, b", + [ + ("http://c:3142", "http://c:3142/"), + ("http://C:3142", "http://c:3142"), + ("http://c", "http://c:80"), + ], + ) + def test_the_same(self, a, b): + assert same_proxy_url(a, b) + + @pytest.mark.parametrize( + "a, b", + [ + ("http://c:3142", "http://c:31420"), + ("http://c:3142", "http://c:3142.evil"), + ("http://c:3142", "https://c:3142"), + ("http://c:3142", ""), + ("http://c:3142", None), + ], + ) + def test_not_the_same(self, a, b): + assert not same_proxy_url(a, b) + + +_REFUSED = """Err:1 http://172.22.8.50:3143/acng-report.html + Could not connect to 172.22.8.50:3143 (172.22.8.50). - connect (111: Connection refused) +E: Failed to fetch http://172.22.8.50:3143/acng-report.html Could not connect to 172.22.8.50:3143 (172.22.8.50). - connect (111: Connection refused) +E: Download Failed""" + +_HTTP_ERROR = """Err:1 http://172.22.8.50:3142/nothing-here.html + 503 Host not found [IP: 172.22.8.50 3142] +E: Failed to fetch http://172.22.8.50:3142/nothing-here.html 503 Host not found [IP: 172.22.8.50 3142] +E: Download Failed""" + +_UNRESOLVED = """Err:1 http://no-such-host.invalid:3142/acng-report.html + Could not resolve 'no-such-host.invalid' +E: Failed to fetch http://no-such-host.invalid:3142/acng-report.html Could not resolve 'no-such-host.invalid' +E: Download Failed""" + +_FETCHED = """Get:1 http://172.22.8.50:3142/acng-report.html +Fetched 14.0 kB in 0s (0 B/s)""" + + +def _fetch(output, rc, match=None): + lines = ["APROXY_BEGIN", "[fetch]", *output.splitlines(), "[rc]", str(rc)] + if match is not None: + lines += ["[match]", str(match)] + return "\n".join([*lines, "APROXY_END"]) + + +class TestFetch: + def test_fetched_and_expected(self): + assert parse_apt_fetch(_fetch(_FETCHED, 0, match=4)) == { + "reachable": True, + "expected": True, + "error": "", + } + + def test_fetched_but_not_the_expected_page(self): + assert parse_apt_fetch(_fetch(_FETCHED, 0, match=0))["expected"] is False + + def test_without_an_expectation_expected_is_none(self): + assert parse_apt_fetch(_fetch(_FETCHED, 0))["expected"] is None + + def test_refused(self): + result = parse_apt_fetch(_fetch(_REFUSED, 100)) + + assert result["reachable"] is False + assert result["error"] == ( + "Could not connect to 172.22.8.50:3143 (172.22.8.50). - connect (111: Connection refused)" + ) + + def test_unresolved(self): + result = parse_apt_fetch(_fetch(_UNRESOLVED, 100)) + assert (result["reachable"], result["error"]) == ( + False, + "Could not resolve 'no-such-host.invalid'", + ) + + def test_an_http_error_was_an_answer(self): + result = parse_apt_fetch(_fetch(_HTTP_ERROR, 100)) + + assert result["reachable"] is True + assert result["expected"] is False + assert result["error"] == "503 Host not found [IP: 172.22.8.50 3142]" + + def test_without_apt_helper_nothing_could_be_tried(self): + with pytest.raises(ValueError, match="apt-helper"): + parse_apt_fetch(_fetch("sh: 1: /usr/lib/apt/apt-helper: not found", 127)) + + def test_no_intact_report_raises(self): + with pytest.raises(ValueError): + parse_apt_fetch("APROXY_BEGIN\n[fetch]\n") + + +class TestFetchCommand: + def test_past_the_proxy_to_the_url_itself(self): + command = apt_fetch_command( + "http://c:3142/acng-report.html", direct=True, expect="Cache NG" + ) + + assert "'Acquire::http::Proxy::c=DIRECT'" in command + assert "grep -c -F -- 'Cache NG'" in command + + def test_through_apt_s_own_proxy(self): + command = apt_fetch_command("http://archive.ubuntu.com/ubuntu/dists/noble/InRelease") + assert "DIRECT" not in command + + @pytest.mark.parametrize("url", ["http://c:3142/x'; rm -rf /", "ftp://c/x", "c:3142"]) + def test_a_url_that_would_break_the_command_is_refused(self, url): + with pytest.raises(ValueError): + apt_fetch_command(url) + + +def test_the_commands_are_valid_posix_sh(): + for command in ( + APT_PROXY_COMMAND, + apt_fetch_command("http://c:3142/x", direct=True, expect="y"), + ): + subprocess.run(["sh", "-n", "-c", command], check=True) + + +@pytest.mark.skipif(not os.path.exists("/usr/bin/apt-config"), reason="needs an apt host") +def test_the_report_command_on_this_host(): + output = subprocess.run(["sh", "-c", APT_PROXY_COMMAND], capture_output=True, text=True).stdout + report = parse_apt_proxy(output) + assert set(report) >= {"proxy", "settings", "https_sources"} + + +class _Host(AptProxyMixin): + """A Debian host: its apt files, and what apt-config would make of them.""" + + def __init__(self, files=None, fetch_rc=0, apt=True): + self.files = dict(files or {}) + self.fetch_rc = fetch_rc + self.apt = apt + self.root = [] + self.fetched = [] + + def _is_apt_host(self): + return self.apt + + def _settings(self): + out = [] + for path in sorted(p for p in self.files if p.startswith(_D)) + [ + p for p in self.files if p == "/etc/apt/apt.conf" + ]: + for n, text in enumerate(self.files[path].splitlines(), 1): + if "proxy" in text.lower(): + out.append((path, n, text)) + return out + + def _effective(self): + value = None + for _path, _n, text in self._settings(): + m = re.match(r'\s*Acquire::http::Proxy\s+"([^"]*)"', text, re.I) + if m: + value = m.group(1) + return value + + def _run_apt_command(self, command): + if command == APT_PROXY_COMMAND: + value = self._effective() + return _report( + dump=[f'Acquire::http::Proxy "{value}";'] if value else [], + parts=[f"{p}:{n}:{t}" for p, n, t in self._settings() if p.startswith(_D)], + main=[f"{p}:{n}:{t}" for p, n, t in self._settings() if not p.startswith(_D)], + sources=["http://deb.debian.org/debian/dists/trixie/"], + ) + self.fetched.append(command) + return _fetch(_FETCHED if self.fetch_rc == 0 else _REFUSED, self.fetch_rc, match=1) + + def _write_apt_file(self, path, content): + self.files[path] = content + return {"ok": True, "output": f"wrote {path}"} + + def _run_apt_root_command(self, command, done): + self.root.append(command) + path = command.rsplit(" ", 1)[-1].strip("'") + for m in re.finditer(r"(\d+)\{/[^}]*\}", command): + n = int(m.group(1)) + lines = self.files[path].splitlines() + lines[n - 1] = "// " + lines[n - 1] + self.files[path] = "\n".join(lines) + "\n" + if command.startswith("rm -f "): + self.files.pop(command.split()[-1].strip("'"), None) + return {"ok": True, "output": done} + + +class TestTheMixin: + def test_it_is_a_package_proxy(self): + assert issubclass(AptProxyMixin, PackageProxyMixin) + + def test_the_fact(self): + host = _Host({f"{_D}00proxy": 'Acquire::http::Proxy "http://c:3142";\n'}) + + assert host.get_package_proxy() == "http://c:3142" + assert host.get_package_proxy_report()["settings"][0]["file"] == f"{_D}00proxy" + + def test_not_an_apt_host(self): + host = _Host(apt=False) + + assert host.get_package_proxy() is None + assert host.get_package_proxy_report() is None + + def test_check_fetches_the_page_past_any_proxy(self): + host = _Host() + + result = host.check_package_proxy("http://c:3142/acng-report.html", expect="Cache") + + assert result == {"reachable": True, "expected": True, "error": ""} + assert "Proxy::c=DIRECT" in host.fetched[0] + + +class TestConfigurePackageProxy: + def test_a_fresh_host(self): + host = _Host() + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is True + assert [s["step"] for s in result["steps"]] == ["configure", "conflicts", "verify", "fetch"] + assert host.files[f"{_D}00proxy"] == 'Acquire::http::Proxy "http://c:3142";\n' + assert "InRelease" in host.fetched[0] and "DIRECT" not in host.fetched[0] + + def test_a_later_file_is_commented_out_and_named(self): + host = _Host( + { + f"{_D}90curtin-aptproxy": 'Acquire::http::Proxy "http://old:3128";\n', + "/etc/apt/apt.conf": 'APT::Foo "1";\nAcquire::http::Proxy "http://older:8080";\n', + } + ) + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is True + assert host.files[f"{_D}90curtin-aptproxy"].startswith("// Acquire::http::Proxy") + assert host.files["/etc/apt/apt.conf"] == ( + 'APT::Foo "1";\n// Acquire::http::Proxy "http://older:8080";\n' + ) + conflicts = result["steps"][1]["output"] + assert f"{_D}90curtin-aptproxy:1 (http://old:3128)" in conflicts + assert "/etc/apt/apt.conf:2 (http://older:8080)" in conflicts + + def test_the_same_proxy_elsewhere_stays(self): + host = _Host({f"{_D}01proxy": 'Acquire::http::Proxy "http://c:3142/";\n'}) + + host.configure_package_proxy("http://c:3142") + + assert host.root == [] + + def test_a_setting_it_cannot_comment_out_fails_verification(self): + host = _Host({"/etc/apt/apt.conf": 'Acquire::http { Proxy "http://older:8080"; };\n'}) + host._effective = lambda: "http://older:8080" + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is False + assert result["steps"][-1]["step"] == "verify" + assert "http://older:8080" in result["steps"][-1]["output"] + assert "/etc/apt/apt.conf:1" in result["steps"][-1]["output"] + + def test_a_dead_proxy_fails_the_fetch(self): + host = _Host(fetch_rc=100) + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is False + assert result["steps"][-1]["step"] == "fetch" + assert "Connection refused" in result["steps"][-1]["output"] + + def test_without_an_http_source_the_fetch_says_so(self): + host = _Host() + real = host._run_apt_command + + def no_sources(command): + return real(command).replace("http://deb.debian.org/debian/dists/trixie/\n", "") + + host._run_apt_command = no_sources + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is True + assert "no http source" in result["steps"][-1]["output"] + + def test_a_direct_source_is_not_proof(self): + host = _Host() + real = host._run_apt_command + host._run_apt_command = lambda command: real(command).replace( + "[parts]", 'Acquire::http::Proxy::deb.debian.org "DIRECT";\n[parts]' + ) + + result = host.configure_package_proxy("http://c:3142") + + assert "no http source" in result["steps"][-1]["output"] + + @pytest.mark.parametrize("url", ['http://x"; Acquire::x "y', "ftp://c", "c:3142"]) + def test_a_url_that_would_break_the_file_is_refused(self, url): + host = _Host() + + result = host.configure_package_proxy(url) + + assert result["success"] is False + assert host.files == {} + + def test_an_empty_url_removes_the_file(self): + host = _Host({f"{_D}00proxy": 'Acquire::http::Proxy "http://c:3142";\n'}) + + result = host.configure_package_proxy("") + + assert result["success"] is True + assert host.files == {} + + def test_only_an_apt_host_takes_one(self): + host = _Host(apt=False) + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is False + assert host.files == {} + + +class TestWhatTheParserSkips: + def test_a_line_that_is_no_grep_output(self): + assert ( + parse_apt_proxy(_report(parts=["grep: /etc/apt/apt.conf.d/x: Permission denied"]))[ + "settings" + ] + == [] + ) + + def test_a_dump_line_of_another_key(self): + report = parse_apt_proxy(_report(dump=['Acquire::http::Timeout "10";'])) + assert report["proxy"] is None + + def test_a_fetch_report_without_an_exit_status_raises(self): + with pytest.raises(ValueError, match="exit status"): + parse_apt_fetch("APROXY_BEGIN\n[fetch]\nGet:1 x\nAPROXY_END\n") + + +class _Plain(AptProxyMixin): + """A driver that says nothing about its package manager: apt is assumed.""" + + +def test_without_its_own_answer_a_host_is_an_apt_host(): + assert _Plain()._is_apt_host() is True + + +class TestNotAnAptHost: + def test_no_check(self): + with pytest.raises(NotImplementedError): + _Host(apt=False).check_package_proxy("http://c:3142/x") + + def test_no_removal(self): + result = _Host(apt=False).configure_package_proxy("") + assert result["success"] is False + assert "not an apt host" in result["output"] + + +def test_a_comment_out_that_fails_stops_the_fix(): + host = _Host({f"{_D}90curtin-aptproxy": 'Acquire::http::Proxy "http://old:3128";\n'}) + host._run_apt_root_command = lambda command, done: {"ok": False, "output": "sed: denied"} + + result = host.configure_package_proxy("http://c:3142") + + assert result["success"] is False + assert result["steps"][-1] == {"step": "conflicts", "ok": False, "output": "sed: denied"} + + +def test_the_comment_out_command(): + assert comment_out_command("/etc/apt/apt.conf", [2, 4]) == ( + "sed -i -e '2{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}' " + "-e '4{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}' /etc/apt/apt.conf" + )