Merge pull request 'feat: apt's proxy on a Proxmox VE node (NetOrk/netork#1397)' (#31) from feat/apt-proxy-verify into master

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

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

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)):

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

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": ""}