From 8436013dcd70b1ac336969bff43fa67a320d179c Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 8 Jul 2026 17:09:19 +0200 Subject: [PATCH] fix(linux): fix_snmp installs snmpd without refreshing apt cache first MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fresh cloud image's apt cache is stale/effectively empty — installing snmpd without an apt-get update first could fail outright or hang on unreachable mirrors, and the install call wasn't guarded, so a timeout propagated as an opaque unguarded exception instead of a clean failure result. Also adds a new apt_update_upgrade device action (apt-get update + upgrade), used by netork's VM-provisioning bootstrap alongside the existing fix_apt_proxy action to fully prep a freshly provisioned VM's apt before installing anything on it. --- napalm_linux/linux.py | 53 +++++++++++++++- tests/test_linux.py | 139 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 2 deletions(-) diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 9fb530f..d1a760a 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -1884,8 +1884,43 @@ class LinuxDriver(OSDriver): return self._action_fix_snmp() if action == "fix_apt_proxy": return self._action_fix_apt_proxy() + if action == "apt_update_upgrade": + return self._action_apt_update_upgrade() raise NotImplementedError(f"Unknown action: {action!r}") + def _action_apt_update_upgrade(self) -> DeviceActionResultDict: + """Refresh the apt cache and upgrade all packages (apt-based systems only).""" + if self._pkg_manager != "apt": + return { + "success": True, + "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.", + } + + sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__") + if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower(): + if not self._sudo_password: + return { + "success": False, + "output": ( + "sudo requires a password on this device but none is configured in " + "netOrk. Please add the sudo password to a Credential Profile assigned " + "to this device, or configure passwordless sudo (NOPASSWD) for this user." + ), + } + + lines: list[str] = [] + try: + out = self._sudo("apt-get update -y 2>&1", read_timeout=90) + lines.append(f"[update] {out.strip()[-300:]}") + out = self._sudo( + "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1", read_timeout=240 + ) + lines.append(f"[upgrade] {out.strip()[-300:]}") + return {"success": True, "output": "\n".join(lines)} + except Exception as exc: + lines.append(f"[error] {exc}") + return {"success": False, "output": "\n".join(lines)} + def _action_fix_snmp(self) -> DeviceActionResultDict: """Install, configure and start snmpd with community 'public'.""" lines: list[str] = [] @@ -1908,6 +1943,16 @@ class LinuxDriver(OSDriver): if not pkg_mgr: return {"success": False, "output": "Package manager not detected — cannot install snmpd."} + # Refresh the package index first — a freshly provisioned (or simply + # long-untouched) system's cache can be stale/empty, which makes the + # install below fail outright rather than just being slow. + if pkg_mgr == "apt": + try: + update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90) + lines.append(f"[update] {update_out.strip()[-200:]}") + except Exception as exc: + lines.append(f"[warn] apt-get update failed: {exc}") + # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing) install_cmd: dict[str, str] = { "apt": "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1", @@ -1918,8 +1963,12 @@ class LinuxDriver(OSDriver): } cmd = install_cmd.get(pkg_mgr) if cmd: - out = self._sudo(cmd, read_timeout=120) - lines.append(f"[install] {out.strip()[-200:]}") + try: + out = self._sudo(cmd, read_timeout=120) + lines.append(f"[install] {out.strip()[-200:]}") + except Exception as exc: + lines.append(f"[error] install failed: {exc}") + return {"success": False, "output": "\n".join(lines)} # 2. Determine the IP netOrk is connecting from by checking the established SSH connection netork_ip = "" diff --git a/tests/test_linux.py b/tests/test_linux.py index 67572c4..3057d3c 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -661,3 +661,142 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver): patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0"]): facts = driver.get_facts() assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute + + +# --------------------------------------------------------------------------- +# _action_fix_snmp / _action_apt_update_upgrade +# --------------------------------------------------------------------------- + + +def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str: + """Canned responses covering every _send() call _action_fix_snmp makes.""" + if command.startswith("sudo -n true"): + return "" # passwordless sudo works + if command.startswith("command -v apt"): + return "/usr/bin/apt" + if command.startswith("command -v"): + return "" # ufw/iptables not found, other pkg managers not found + if command.startswith("ss -tnp"): + return "" # no netork_ip detected — skips firewall step + if command.startswith("cat /etc/snmp/snmpd.conf"): + return "agentAddress udp:161\nrocommunity public\n" + if command.startswith("snmpget"): + return "STRING: Linux test" + return "" + + +class TestActionFixSnmp: + def test_runs_apt_get_update_before_install(self, driver): + driver._pkg_manager = "apt" + driver._sudo_password = None + sudo_calls: list[str] = [] + + def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: + sudo_calls.append(command) + return "" + + with ( + patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), + patch.object(driver, "_sudo", side_effect=_sudo_side_effect), + ): + driver._action_fix_snmp() + + update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c) + install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c) + assert update_idx < install_idx + + def test_install_exception_returns_failure_instead_of_raising(self, driver): + driver._pkg_manager = "apt" + driver._sudo_password = None + + def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: + if "apt-get install" in command: + raise TimeoutError("connection timed out") + return "" + + with ( + patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), + patch.object(driver, "_sudo", side_effect=_sudo_side_effect), + ): + result = driver._action_fix_snmp() + + assert result["success"] is False + assert "install failed" in result["output"] + + def test_apt_get_update_failure_is_non_fatal(self, driver): + """apt-get update failing (e.g. transient network issue) must not + abort the whole action — install is still attempted.""" + driver._pkg_manager = "apt" + driver._sudo_password = None + + def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: + if "apt-get update" in command: + raise TimeoutError("network unreachable") + return "" + + with ( + patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), + patch.object(driver, "_sudo", side_effect=_sudo_side_effect), + ): + result = driver._action_fix_snmp() + + assert "apt-get update failed" in result["output"] + + +class TestActionAptUpdateUpgrade: + def test_skips_when_not_apt(self, driver): + driver._pkg_manager = "dnf" + result = driver._action_apt_update_upgrade() + assert result["success"] is True + assert "Skipped" in result["output"] + + def test_fails_when_sudo_needs_password_and_none_configured(self, driver): + driver._pkg_manager = "apt" + driver._sudo_password = None + with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"): + result = driver._action_apt_update_upgrade() + assert result["success"] is False + assert "sudo requires a password" in result["output"] + + def test_runs_update_then_upgrade_successfully(self, driver): + driver._pkg_manager = "apt" + driver._sudo_password = "secret" # noqa: S105 + with ( + patch.object(driver, "_send", return_value=""), + patch.object( + driver, + "_sudo", + side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"], + ) as mock_sudo, + ): + result = driver._action_apt_update_upgrade() + + assert result["success"] is True + assert "[update]" in result["output"] + assert "[upgrade]" in result["output"] + update_call, upgrade_call = mock_sudo.call_args_list + assert "apt-get update" in update_call.args[0] + assert "apt-get upgrade" in upgrade_call.args[0] + + def test_exception_during_upgrade_returns_failure(self, driver): + driver._pkg_manager = "apt" + driver._sudo_password = "secret" # noqa: S105 + with ( + patch.object(driver, "_send", return_value=""), + patch.object( + driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")] + ), + ): + result = driver._action_apt_update_upgrade() + + assert result["success"] is False + assert "[error]" in result["output"] + + +class TestRunDeviceActionDispatch: + def test_apt_update_upgrade_action_dispatches(self, driver): + with patch.object( + driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""} + ) as mock_action: + driver.run_device_action("apt_update_upgrade") + mock_action.assert_called_once()