From 04f071afd044158cac07c83c25a08bec5bbfb952 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 22 Oct 2026 19:16:40 +0200 Subject: [PATCH] feat: apt's proxy through 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. The driver now supplies only the transports: reads on the exec channel, root commands through _root_outcome, the file through the staged _install_file_as_root. New: get_package_proxy_report() and check_package_proxy(). configure_package_proxy() also comments out other proxy lines, reads the result back and fetches through the proxy. Requires napalm-device-types 4.22.0; version 0.14.0. --- CHANGELOG.md | 1 + napalm_linux/linux.py | 43 +++++++++------------ pyproject.toml | 4 +- tests/test_fix_templates.py | 76 +++++++++++++++++++++++++++++++++++-- 4 files changed, 93 insertions(+), 31 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c858b9a..73c2648 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 7d5f827..8538341 100644 --- a/napalm_linux/linux.py +++ b/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.""" diff --git a/pyproject.toml b/pyproject.toml index ea47f3c..8ce076e 100644 --- a/pyproject.toml +++ b/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 ] diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py index 51e820c..053ab9c 100644 --- a/tests/test_fix_templates.py +++ b/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"})