Merge pull request 'feat: AptProxyMixin, what apt fetches through and pointing it at a proxy (NetOrk/netork#1397)' (#53) from feat/apt-proxy-verify into main

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.

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,

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",

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::<host>``), 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::<host>``: 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",
+]

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"

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"
+    )