napalm_linux/linux.py

@@ -32,11 +32,17 @@ from netmiko.exceptions import (
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException
 from napalm.base.netmiko_helpers import netmiko_args
 from napalm_device_types import (
+    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
     FingerprintRule,
+    HostStatusMixin,
     KernelFactsMixin,
     OSDriver,
     SystemdServicesMixin,
     SystemdUnavailable,
+    parse_apt_upgradable,
+    parse_dnf_security,
+    strip_terminal_codes,
 )
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
@@ -62,6 +68,33 @@ _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
 _RC_MARKER = "__NETORK_RC="
 _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
 
+
+def _split_status(raw: str) -> tuple[str, int | None]:
+    """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``.
+
+    The status is ``None`` when the marker never arrived (output cut short), so
+    a caller can tell "unknown" from "succeeded".
+    """
+    raw = strip_terminal_codes(raw)
+    matches = list(_RC_MARKER_RE.finditer(raw))
+    if not matches:
+        return raw, None
+    last = matches[-1]
+    return (raw[: last.start()] + raw[last.end():]).strip(), int(last.group(1))
+
+
+#: How each package manager refreshes its index. pacman is left out on purpose:
+#: ``pacman -Sy`` without ``-u`` invites a partial upgrade on the next install.
+_REFRESH = {
+    # No LC_ALL=C here: under sudo it is an environment variable sudoers may refuse
+    # to set. Only the exit status decides, so the language is merely what is shown.
+    "apt": "apt-get update -q 2>&1",
+    "dnf": "dnf makecache -q 2>&1",
+    "yum": "yum makecache -q 2>&1",
+    "apk": "apk update -q 2>&1",
+}
+_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
+
 #: What to do when sudo wants a password netOrk does not have.
 _SUDO_PASSWORD_HINT = (
     "sudo requires a password on this device but none is configured in netOrk. "
@@ -151,7 +184,7 @@ def _short_image_id(raw: str) -> str:
     return raw.strip().removeprefix("sha256:")[:12]
 
 
-class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
+class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -297,13 +330,9 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
         The status is ``None`` when the marker never arrived (output cut short),
         so a caller can tell "unknown" from "succeeded".
         """
-        raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
+        return _split_status(
-        matches = list(_RC_MARKER_RE.finditer(raw))
+            self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
-        if not matches:
+        )
-            return raw, None
-        last = matches[-1]
-        output = (raw[: last.start()] + raw[last.end():]).strip()
-        return output, int(last.group(1))
 
     def _is_root(self) -> bool:
         """Whether the SSH user is root, asked once per session.
@@ -321,12 +350,24 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
         Without a sudo password, ``sudo -n`` fails at once where a prompt would
         otherwise hang the session until the read timeout.
         """
-        if not privileged or self._is_root():
+        if not privileged:
+            return self._send(command, read_timeout=timeout)
+        return self._run_privileged(command, timeout)
+
+    def _run_privileged(self, command: str, timeout: float = 100) -> str:
+        """Run *command* as root: directly for a root login, through ``_sudo``
+        with a sudo password, and through ``sudo -n`` without one -- which fails at
+        once where a password prompt would hang the session until the timeout."""
+        if self._is_root():
             return self._send(command, read_timeout=timeout)
         if self._sudo_password:
             return self._sudo(command, read_timeout=timeout)
         return self._send(f"sudo -n {command}", read_timeout=timeout)
 
+    def _run_host_status_command(self, command: str) -> str:
+        """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
+        return self._send(command, read_timeout=60)
+
     def _detect_pkg_manager(self) -> str | None:
         """Return the first package manager binary found on PATH."""
         for pm in _PKG_MANAGERS:
@@ -1243,48 +1284,49 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
     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.
-        out = self._send(
+        # Raises ValueError when apt failed or the output was cut short.
-            "LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'",
+        return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60))
-            read_timeout=60,
-        )
-        # Join wrapped lines: netmiko's 80-col pseudo-TTY causes long apt lines to
-        # break; continuation lines start with a space.
-        raw_lines: List[str] = []
-        for line in out.splitlines():
-            if line.startswith(" ") and raw_lines:
-                raw_lines[-1] += line.strip()
-            else:
-                raw_lines.append(line)
-        updates: list[UpdateDict] = []
-        for line in raw_lines:
-            # openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]
-            m = re.match(
-                r"^(\S+)/\S+\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]", line
-            )
-            if m:
-                updates.append({
-                    "name": m.group(1),
-                    "current_version": m.group(3),
-                    "new_version": m.group(2),
-                })
-        return updates
 
     def _get_updates_rpm(self) -> list[UpdateDict]:
+        """dnf/yum check-update: exit 100 means updates, 0 none, anything else failed."""
         cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null"
-        out = self._sudo(cmd)
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
+        if status not in (0, 100):
+            raise RuntimeError(f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}")
+        security = self._rpm_security_names()
         updates: list[UpdateDict] = []
-        for line in out.splitlines():
+        for line in output.splitlines():
             parts = line.split()
             if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
-                name_arch = parts[0]
+                name = parts[0].rsplit(".", 1)[0]
-                name = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch
                 updates.append({
                     "name": name,
                     "current_version": "",
                     "new_version": parts[1],
+                    "origin": parts[2] if len(parts) >= 3 else None,
+                    "security": None if security is None else name in security,
                 })
         return updates
 
+    def _rpm_security_names(self) -> set[str] | None:
+        """Packages a pending security advisory covers; None when dnf/yum cannot say."""
+        cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
+        return parse_dnf_security(output) if status == 0 else None
+
+    def refresh_available_updates(self) -> dict[str, Any]:
+        """Refresh the package index (apt-get update, dnf makecache, apk update)."""
+        cmd = _REFRESH.get(self._pkg_manager or "")
+        if cmd is None:
+            return {
+                "success": False,
+                "output": f"Refreshing the index is not supported for {self._pkg_manager!r}",
+            }
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 180))
+        if status != 0 and "password is required" in output:
+            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
+        return {"success": status == 0, "output": output}
+
     def _get_updates_apk(self) -> list[UpdateDict]:
         out = self._send("apk version -l '<' 2>/dev/null")
         updates: list[UpdateDict] = []

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=2.2.0",
+    "napalm-device-types>=2.3.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -226,6 +226,7 @@ APT_UPGRADABLE = (
     "Listing... Done\n"
     "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n"
     "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n"
+    "__APT_RC=0\n"
 )
 
 
@@ -1282,3 +1283,135 @@ class TestManageService:
             driver.manage_service("cron; reboot", "stop")
 
         assert driver._device.send_command.call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# Updates: origin and security, refresh, host status (netOrk MVP 5)
+# ---------------------------------------------------------------------------
+
+APT_WITH_SECURITY = (
+    "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n"
+    "docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]\n"
+    "__APT_RC=0\n"
+)
+
+
+class TestAvailableUpdates:
+    def test_apt_reports_origin_and_security(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, APT_WITH_SECURITY)
+
+        updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        assert updates["openssl"]["security"] is True
+        assert updates["openssl"]["origin"] == "noble-updates,noble-security"
+        assert updates["docker-compose-plugin"]["security"] is False
+
+    def test_apt_that_could_not_read_raises_instead_of_reporting_nothing(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, "E: Could not open lock file\n__APT_RC=100\n")
+
+        with pytest.raises(ValueError):
+            driver.get_available_updates()
+
+    def test_apt_without_an_exit_status_raises(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, "openssl/noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro")
+
+        with pytest.raises(ValueError):
+            driver.get_available_updates()
+
+    def test_dnf_marks_what_a_security_advisory_covers(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = [
+            "0",  # id -u
+            "openssl-libs.x86_64   1:3.1.4-2.fc40   updates\n"
+            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
+            "FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n__NETORK_RC=0",
+        ]
+
+        updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        assert updates["openssl-libs"]["security"] is True
+        assert updates["vim-enhanced"]["security"] is False
+
+    def test_dnf_without_advisories_leaves_security_unknown(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = [
+            "0",
+            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
+            "Error: updateinfo metadata missing\n__NETORK_RC=1",
+        ]
+
+        assert driver.get_available_updates()[0]["security"] is None
+
+    def test_dnf_that_failed_raises(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = ["0", "Error: Failed to download metadata\n__NETORK_RC=1"]
+
+        with pytest.raises(RuntimeError):
+            driver.get_available_updates()
+
+
+class TestRefreshAvailableUpdates:
+    def _sent(self, driver) -> list:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_apt_refreshes_its_index_as_root(self, driver):
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["0", "Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0"]
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is True
+        assert "apt-get update" in self._sent(driver)[1]
+
+    def test_without_a_sudo_password_it_never_waits_for_one(self, driver):
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["1000", "sudo: a password is required\n__NETORK_RC=1"]
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        assert self._sent(driver)[1].startswith("sudo -n apt-get update")
+
+    def test_dnf_refreshes_its_metadata(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = ["0", "Metadata cache created.\n__NETORK_RC=0"]
+
+        assert driver.refresh_available_updates()["success"] is True
+        assert "dnf makecache" in self._sent(driver)[1]
+
+    def test_pacman_is_not_refreshed_on_its_own(self, driver):
+        """pacman -Sy without -u invites a partial upgrade on the next install."""
+        driver._pkg_manager = "pacman"
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        driver._device.send_command.assert_not_called()
+
+
+class TestHostStatus:
+    def test_the_driver_carries_the_shared_command(self, driver):
+        from napalm_device_types.host_status import HOST_STATUS_COMMAND
+
+        _mock_send(
+            driver,
+            "HSTAT_BEGIN\n[reboot-required]\n[kernel]\n6.8.0-142-generic\n[modules]\n"
+            "6.8.0-142-generic\n[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n",
+        )
+
+        status = driver.get_host_status()
+
+        assert driver._device.send_command.call_args[0][0] == HOST_STATUS_COMMAND
+        assert status["reboot_required"] is True
+
+
+class TestTerminalCodes:
+    def test_a_status_marker_behind_a_terminal_code_is_still_read(self, driver):
+        """apt-get on a pseudo-terminal leaves keypad codes in front of the marker."""
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"]
+
+        assert driver.refresh_available_updates()["success"] is True