feat: AptProxyMixin, what apt fetches through and pointing it at a proxy (NetOrk/netork#1397)

apt reads 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
90curtin-aptproxy: reading 00proxy alone missed both, and a fix that wrote
00proxy and read it back reported success while apt fetched elsewhere.

- get_package_proxy_report(): apt-config dump's proxy, per-host overrides,
  auto-detect, every line of apt's files that sets it (in reading order,
  editable or inside a block), the http and https sources.
- check_package_proxy(url, expect): a fetch with apt-helper, past any proxy;
  an HTTP error counts as an answer.
- configure_package_proxy(url): write 00proxy, comment out other proxy
  lines, read apt's proxy back, fetch one InRelease through it.
- same_proxy_url(): scheme, host and port, a trailing slash aside.

A driver supplies _run_apt_command, _run_apt_root_command and
_write_apt_file. Version 4.22.0.

napalm_device_types/__init__.py

@@ -153,6 +153,13 @@ from napalm_device_types.phone import PhoneDriver
 from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping
 from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of
 from napalm_device_types.host_status import HOST_STATUS_COMMAND, HostStatusMixin, parse_host_status
+from napalm_device_types.apt_proxy import (
+    APT_PROXY_COMMAND,
+    AptProxyMixin,
+    parse_apt_fetch,
+    parse_apt_proxy,
+    same_proxy_url,
+)
 from napalm_device_types.package_updates import (
     APT_UPGRADABLE_COMMAND,
     DNF_SECURITY_COMMAND,
@@ -221,6 +228,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,404 @@
+# -*- coding: utf-8 -*-
+"""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 Dict, List, Optional, TYPE_CHECKING
+from urllib.parse import urlsplit
+
+from napalm_device_types.fixes import PackageProxyMixin, run_fix_steps
+from napalm_device_types.models import (
+    AptProxyReportDict,
+    AptProxySettingDict,
+    FixResultDict,
+    FixStepOutcomeDict,
+    ProxyFetchDict,
+)
+from napalm_device_types.terminal import strip_terminal_codes
+
+_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]):
+        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) -> Optional[AptProxySettingDict]:
+    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 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: Optional[str] = None
+    per_host: Dict[str, str] = {}
+    auto_detect: Optional[str] = None
+    for line in sections.get("dump", []):
+        found = _DUMP.match(line.strip())
+        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
+    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://")],
+    }
+
+
+def same_proxy_url(a: Optional[str], b: Optional[str]) -> bool:
+    """Whether *a* and *b* name the same proxy: scheme, host and port, a trailing ``/`` aside."""
+    if not a or not b:
+        return False
+    try:
+        x, y = urlsplit(a), urlsplit(b)
+        ports = (x.port or _DEFAULT_PORT.get(x.scheme), y.port or _DEFAULT_PORT.get(y.scheme))
+    except ValueError:
+        return False
+    return (
+        x.scheme == y.scheme
+        and (x.hostname or "") == (y.hostname or "")
+        and ports[0] == ports[1]
+        and x.path.rstrip("/") == y.path.rstrip("/")
+    )
+
+
+_DEFAULT_PORT = {"http": 80, "https": 443}
+
+
+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}
+
+
+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) -> Optional[AptProxyReportDict]:
+        """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) -> Optional[str]:
+        """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:
+        report = self.get_package_proxy_report()
+        if report is None:
+            raise ValueError("not an apt host")
+        return report
+
+    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)
+        ]
+        by_file: Dict[str, List[int]] = {}
+        for s in others:
+            if s["editable"]:
+                by_file.setdefault(s["file"], []).append(s["line"])
+        done: List[str] = []
+        for path, numbers in by_file.items():
+            edits = " ".join(
+                f"-e '{n}{{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}}'"
+                for n in numbers
+            )
+            outcome = self._run_apt_root_command(
+                f"sed -i {edits} {shlex.quote(path)}", f"commented out in {path}"
+            )
+            if not outcome["ok"]:
+                return outcome
+        for s in others:
+            what = "commented out" if s["editable"] else "left alone (inside a block)"
+            done.append(f"{s['file']}:{s['line']} ({s['value']}) {what}")
+        return {"ok": True, "output": "; ".join(done) 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",
+    "PROXY_URL",
+    "apt_fetch_command",
+    "parse_apt_fetch",
+    "parse_apt_proxy",
+    "same_proxy_url",
+]

napalm_device_types/models.py

@@ -120,6 +120,45 @@ class HostStatusDict(TypedDict):
     filesystems: NotRequired[Optional[List[FilesystemUsageDict]]]
 
 
+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: Optional[str]
+    #: ``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: Optional[str]
+    #: 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: Optional[bool]
+    #: apt's reason when the fetch failed, else "".
+    error: str
+
+
 # ---------------------------------------------------------------------------
 # Access Point
 # ---------------------------------------------------------------------------