CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- apt's proxy through napalm-device-types' `AptProxyMixin` (NetOrk/netork#1397). `get_package_proxy()` is what `apt-config dump` says, not `00proxy` alone: a proxy in `90curtin-aptproxy` or `/etc/apt/apt.conf` wins over it, and was missed. New: `get_package_proxy_report()` (every line that sets it, per-host overrides, the https sources) and `check_package_proxy(url, expect)` (a fetch with `apt-helper`, past any proxy). `configure_package_proxy(url)` now also comments out other proxy lines, reads the result back and fetches one `InRelease` through the proxy; it fails when apt still fetches elsewhere. Reads run on the exec channel. Requires napalm-device-types 4.22.0; version 0.14.0.
 - `get_host_firewall()` (napalm-device-types' `HostFirewallMixin`, NetOrk/netork#1344): the host's own netfilter rules — `nft -j list ruleset`, with the tables iptables manages taken from `iptables-save` — as chains and rules in one shape. Read-only, as root, on an exec channel (the nft JSON is one long line). Requires napalm-device-types 4.21.0; version 0.13.0.
 - `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd service is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, on an exec channel under `sh`, without sudo. Requires napalm-device-types 4.20.0; version 0.12.0.
 

napalm_linux/linux.py

@@ -46,7 +46,7 @@ from napalm_device_types import (
     ListeningSocketsMixin,
     OpenSshServerMixin,
     OSDriver,
-    PackageProxyMixin,
+    AptProxyMixin,
     ServiceClustersMixin,
     ServicePackagesMixin,
     SnmpAgentMixin,
@@ -261,8 +261,6 @@ _SNMP_INSTALL = {
     "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
     "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
 }
-_APT_PROXY_CONF = "/etc/apt/apt.conf.d/00proxy"
-_PROXY_URL = re.compile(r"^https?://[A-Za-z0-9.\-\[\]:@%_~/]+$")
 _USER_NAME = re.compile(r"^[a-z_][a-z0-9_.\-]{0,31}$")
 #: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
 _SNMP_WORD = re.compile(r"^[^\s\"#]+$")
@@ -294,7 +292,7 @@ def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
 class LinuxDriver(
     SnmpAgentMixin,
     EnsurePackageMixin,
-    PackageProxyMixin,
+    AptProxyMixin,
     KernelFactsMixin,
     ServicePackagesMixin,
     ListeningSocketsMixin,
@@ -1544,14 +1542,6 @@ class LinuxDriver(
             f"Package manager '{self._pkg_manager}' is not supported"
         )
 
-    def get_package_proxy(self) -> str | None:
-        """The proxy apt fetches through (``00proxy``); None for none or a non-apt host."""
-        if self._pkg_manager != "apt":
-            return None
-        conf = self._send(f"cat {_APT_PROXY_CONF} 2>/dev/null || true")
-        found = re.search(r'Acquire::http::Proxy\s+"([^"]+)"', conf)
-        return found.group(1) if found else None
-
     def _get_updates_apt(self) -> list[UpdateDict]:
         # apt list --upgradable does not need root; avoid sudo so it works even
         # without a configured sudo password.
@@ -2052,22 +2042,23 @@ class LinuxDriver(
             "snmpd enabled and running",
         )
 
-    def configure_package_proxy(self, url: str) -> FixResultDict:
-        """apt fetches through *url* (``/etc/apt/apt.conf.d/00proxy``); ``""`` removes it."""
+    # apt's proxy (napalm-device-types AptProxyMixin): the transports
 
-        def write() -> FixStepOutcomeDict:
-            if self._pkg_manager != "apt":
-                return {"ok": False, "output": f"package manager is {self._pkg_manager!r}, not apt"}
-            if not url:
-                return self._root_outcome(f"rm -f {_APT_PROXY_CONF}", 30, "proxy removed")
-            if not _PROXY_URL.match(url):
-                raise ValueError(f"not a proxy URL: {url!r}")
-            written = self._install_file_as_root(_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n')
-            if written.exit_code:
-                return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
-            return {"ok": True, "output": f"apt fetches through {url}"}
+    def _is_apt_host(self) -> bool:
+        return self._pkg_manager == "apt"
 
-        return run_fix_steps([("configure", write)])
+    def _run_apt_command(self, command: str) -> str:
+        """On the exec channel: no prompt to mistake a line of apt's output for."""
+        return self.run_command(command, timeout=60).stdout
+
+    def _run_apt_root_command(self, command: str, done: str) -> FixStepOutcomeDict:
+        return self._root_outcome(command, 30, done)
+
+    def _write_apt_file(self, path: str, content: str) -> FixStepOutcomeDict:
+        written = self._install_file_as_root(path, content)
+        if written.exit_code:
+            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
+        return {"ok": True, "output": f"wrote {path}"}
 
     def grant_container_engine_access(self, user: str) -> FixResultDict:
         """Add *user* to the ``docker`` group, so the engine answers without sudo."""

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.13.0"
+version = "0.14.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.21.0",
+    "napalm-device-types>=4.22.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_fix_templates.py

@@ -123,21 +123,72 @@ class TestSnmpAgentSteps:
         driver._install_file_as_root.assert_not_called()
 
 
+_REPORT = (
+    "APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n"
+    "http://deb.debian.org/debian/dists/trixie/\nAPROXY_END\n"
+)
+
+
+def _host(driver, proxy="", parts="", fetched=0):
+    """What the host answers on the exec channel: apt's report, then a fetch."""
+    ran: list[str] = []
+
+    def run_command(command, *, privileged=False, timeout=60, stdin=None):
+        ran.append(command)
+        if "apt-config dump" in command:
+            dump = f'Acquire::http::Proxy "{proxy}";\n' if proxy else ""
+            return MagicMock(exit_code=0, stdout=_REPORT.format(dump=dump, parts=parts), stderr="")
+        out = f"APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n{fetched}\nAPROXY_END\n"
+        return MagicMock(exit_code=0, stdout=out, stderr="")
+
+    driver.run_command = run_command
+    return ran
+
+
 class TestPackageProxyAndEngineAccess:
     def test_the_proxy_is_the_caller_s(self, driver):
         driver._install_file_as_root = MagicMock(
             return_value=MagicMock(exit_code=0, stdout="", stderr="")
         )
+        _host(driver, proxy="http://cache.lan:3142")
 
         result = driver.configure_package_proxy("http://cache.lan:3142")
 
-        assert result["success"] is True
+        assert result["success"] is True, result["output"]
         path, content = driver._install_file_as_root.call_args.args
         assert (path, content) == (
             "/etc/apt/apt.conf.d/00proxy",
             'Acquire::http::Proxy "http://cache.lan:3142";\n',
         )
 
+    def test_it_reads_back_and_fetches_through_it(self, driver):
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+        ran = _host(driver, proxy="http://cache.lan:3142")
+
+        result = driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert [s["step"] for s in result["steps"]] == ["configure", "conflicts", "verify", "fetch"]
+        assert any("dists/trixie/InRelease" in c for c in ran)
+
+    def test_another_file_s_proxy_is_commented_out_as_root(self, driver):
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+        _host(
+            driver,
+            proxy="http://cache.lan:3142",
+            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
+        )
+        ran = []
+        driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
+
+        driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert ran and ran[0].startswith("sed -i -e '1{")
+        assert ran[0].endswith("/etc/apt/apt.conf.d/90curtin-aptproxy")
+
     def test_a_proxy_url_that_would_break_the_file_is_refused(self, driver):
         driver._install_file_as_root = MagicMock()
 
@@ -151,6 +202,15 @@ class TestPackageProxyAndEngineAccess:
 
         assert driver.configure_package_proxy("http://cache.lan:3142")["success"] is False
 
+    def test_the_proxy_answers_past_apt_s_proxy(self, driver):
+        out = "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n3\nAPROXY_END\n"
+        driver.run_command = MagicMock(return_value=MagicMock(exit_code=0, stdout=out, stderr=""))
+
+        result = driver.check_package_proxy("http://c:3142/acng-report.html", expect="Cache")
+
+        assert result == {"reachable": True, "expected": True, "error": ""}
+        assert "Proxy::c=DIRECT" in driver.run_command.call_args.args[0]
+
     def test_engine_access_for_the_user_the_caller_names(self, driver):
         ran = []
         driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
@@ -220,16 +280,26 @@ class TestFactsNotWarnings:
         assert "get_device_warnings" not in vars(LinuxDriver)
 
     def test_the_package_proxy_apt_fetches_through(self, driver):
-        driver._send = lambda command, read_timeout=100: 'Acquire::http::Proxy "http://c:3142";'
+        _host(driver, proxy="http://c:3142")
         assert driver.get_package_proxy() == "http://c:3142"
 
     def test_no_proxy_set(self, driver):
-        driver._send = lambda command, read_timeout=100: ""
+        _host(driver)
         assert driver.get_package_proxy() is None
 
+    def test_where_apt_s_proxy_is_set(self, driver):
+        _host(
+            driver,
+            proxy="http://old:3128",
+            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
+        )
+        report = driver.get_package_proxy_report()
+        assert report["settings"][0]["file"] == "/etc/apt/apt.conf.d/90curtin-aptproxy"
+
     def test_not_an_apt_host(self, driver):
         driver._pkg_manager = "dnf"
         assert driver.get_package_proxy() is None
+        assert driver.get_package_proxy_report() is None
 
     def test_the_applications_proxy_url_is_no_argument_any_more(self):
         d = LinuxDriver("h", "admin", "p", optional_args={"apt_proxy_url": "http://c:3142"})