diff --git a/CHANGELOG.md b/CHANGELOG.md index 7381aa5..7dc9fd7 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 (napalm-device-types' `AptProxyMixin`, NetOrk/netork#1397): `get_package_proxy()`, `get_package_proxy_report()`, `check_package_proxy(url, expect)` and the fix `configure_package_proxy(url)`, as on any Debian host. The node is logged in as root, so the file is written and other proxy lines are commented out over the exec path. Requires napalm-device-types 4.22.0; version 0.7.0. - `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd unit of the node is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, over the exec path. Requires napalm-device-types 4.20.0; version 0.6.0. ### Security diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 0afa11e..fdff1ec 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -35,6 +35,7 @@ from typing import Any logger = logging.getLogger(__name__) from napalm_device_types import ( + AptProxyMixin, EnsurePackageMixin, FingerprintRule, HostStatusMixin, @@ -99,6 +100,7 @@ class ProxmoxDriver( LicenseMixin, SnmpAgentMixin, EnsurePackageMixin, + AptProxyMixin, HypervisorDriver, ): """NAPALM driver for Proxmox VE nodes.""" diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py index 21a893e..ce962ef 100644 --- a/napalm_proxmox/system_mixin.py +++ b/napalm_proxmox/system_mixin.py @@ -20,6 +20,7 @@ import base64 import ipaddress import logging import re +import shlex from typing import Any from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable @@ -512,6 +513,25 @@ class ProxmoxSystemMixin: output, status = self._rc(command) return {"ok": status == 0, "output": done if status == 0 else output[-300:]} + # apt's proxy (napalm-device-types AptProxyMixin): the transports. The node + # is logged in as root, so every command runs as root. + + def _run_apt_command(self, command: str) -> str: + return str(self._exec_ssh_command(command)) + + def _run_apt_root_command(self, command: str, done: str) -> dict: + return self._step(command, done) + + def _write_apt_file(self, path: str, content: str) -> dict: + encoded = base64.b64encode(content.encode()).decode() + # No ``exit`` here: ``_rc`` reads the status after the command's braces. + return self._step( + f"umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > \"$t\" && " + f"install -m 644 -o root -g root \"$t\" {shlex.quote(path)}; " + "s=$?; rm -f \"$t\"; [ \"$s\" -eq 0 ]", + f"wrote {path}", + ) + def install_package(self, name: str, version: str = "") -> None: """apt-get install *name* (``PackageManagementMixin``); raises when apt fails.""" if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)): diff --git a/pyproject.toml b/pyproject.toml index 66aa320..a4b7e63 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-proxmox" -version = "0.6.0" +version = "0.7.0" description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)" readme = "README.md" license = { text = "Apache-2.0" } @@ -25,7 +25,7 @@ classifiers = [ requires-python = ">=3.9" dependencies = [ "napalm>=5.0.0", - "napalm_device_types>=4.20.0", + "napalm_device_types>=4.22.0", "paramiko>=5.0.0", # CVE-2026-44405; imported directly for SSH fallback (driver.py) "proxmoxer>=2.0.0", "netaddr>=0.9.0", diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py index dfaa5af..f72021f 100644 --- a/tests/test_fix_templates.py +++ b/tests/test_fix_templates.py @@ -116,3 +116,77 @@ class TestPackages: def test_the_old_actions_are_gone(): for name in ("run_device_action", "_action_fix_snmp", "_action_install_lldpd"): assert not hasattr(ProxmoxDriver, name), name + + +_APT_REPORT = ( + "APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n" + "http://download.proxmox.com/debian/pve/dists/trixie/\nAPROXY_END\n" +) + + +class _AptNode(_Node): + """A node that answers apt's proxy report and a fetch.""" + + def __init__(self, proxy="", parts=""): + super().__init__() + self.proxy = proxy + self.parts = parts + + def __call__(self, command: str) -> str: + if "apt-config dump" in command: + self.ran.append(command) + dump = f'Acquire::http::Proxy "{self.proxy}";\n' if self.proxy else "" + return _APT_REPORT.format(dump=dump, parts=self.parts) + if "apt-helper" in command: + self.ran.append(command) + return "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n2\nAPROXY_END\n" + return super().__call__(command) + + +class TestPackageProxy: + """Proxmox VE is Debian: apt's proxy is set and read as on any Debian host (netOrk #1397).""" + + def test_it_takes_the_fix(self): + assert callable(getattr(ProxmoxDriver, "configure_package_proxy", None)) + + def test_the_fact(self): + d = ProxmoxDriver("pve", "root@pam", "pw") + d._exec_ssh_command = _AptNode(proxy="http://c:3142") + + assert d.get_package_proxy() == "http://c:3142" + + def test_written_as_root_read_back_and_fetched_through(self): + node = _AptNode(proxy="http://c:3142") + d = ProxmoxDriver("pve", "root@pam", "pw") + d._exec_ssh_command = node + + result = d.configure_package_proxy("http://c:3142") + + assert result["success"] is True, result["output"] + write = next(c for c in node.ran if "/etc/apt/apt.conf.d/00proxy" in c) + assert "install -m 644" in write + b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1) + import base64 + + assert base64.b64decode(b64).decode() == 'Acquire::http::Proxy "http://c:3142";\n' + assert any("dists/trixie/InRelease" in c for c in node.ran) + + def test_another_proxy_is_commented_out(self): + node = _AptNode( + proxy="http://c:3142", + parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n', + ) + d = ProxmoxDriver("pve", "root@pam", "pw") + d._exec_ssh_command = node + + d.configure_package_proxy("http://c:3142") + + assert any(c.startswith("{ sed -i -e '1{") for c in node.ran) + + def test_the_proxy_answers(self): + d = ProxmoxDriver("pve", "root@pam", "pw") + d._exec_ssh_command = _AptNode() + + result = d.check_package_proxy("http://c:3142/acng-report.html", expect="Cache") + + assert result == {"reachable": True, "expected": True, "error": ""}