fix: adapt to napalm-device-types 4.7
CI / test (3.10) (push) Successful in 38s
CI / test (3.11) (push) Successful in 37s
CI / test (3.12) (push) Successful in 39s

napalm-device-types 4.7 changed three things this driver depends on:

- parse_dnf_security needs the "__DNF_RC=<status>" line that
  DNF_SECURITY_COMMAND prints and raises ValueError without it or when dnf
  failed. The yum security query had no such line, so listing updates on
  any yum host raised. _YUM_SECURITY_COMMAND now prints the marker the same
  way ("{ ...; echo __DNF_RC=$?; } | cat"). _rpm_security_names checks
  __DNF_RC itself and turns a failed query, or output cut short before the
  marker, into None (security unknown) rather than "no advisories" or an
  exception.
- OSDriver.get_users was renamed to get_os_users, since NAPALM's own
  get_users has another shape. LinuxDriver.get_users is now get_os_users
  with the same return value.
- run_on_transport reads an exec channel to EOF after its exit status
  arrives; the tests' fake channels gain recv/recv_stderr for that.

Require napalm-device-types>=4.7.1 and bump the version to 0.6.0.
This commit is contained in:
christianmanivong committed 2026-10-13 06:25:17 +00:00
1 parent 81b2df9d71
commit 350e3a6d6e
8 files changed
+141 -15

No files matched your search

+18
View File
@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [Unreleased]
### Changed
- Requires napalm-device-types 4.7.1 or later; version 0.6.0.
- **Breaking:** `get_users()` is now `get_os_users()`, following napalm-device-types
4.7, which renamed `OSDriver.get_users` because NAPALM's own `get_users` has
another shape. The return value is unchanged (a list of `UserDict`).
`get_users()` is NAPALM's again and raises `NotImplementedError`.
### Fixed
- Listing updates on yum hosts (RHEL 7, CentOS 7) no longer fails with
napalm-device-types 4.7: its `parse_dnf_security` needs the `__DNF_RC=<status>`
line `DNF_SECURITY_COMMAND` prints, and the yum security query had none, so it
raised `ValueError`. The yum query now prints that marker the same way
(`{ ...; echo __DNF_RC=$?; } | cat`). A yum or dnf security query that failed,
or whose output was cut short before the marker, still leaves `security` unknown
(`None`) instead of reporting no advisories.
### Security ### Security
- `run_command(privileged=True, stdin=...)` raises `ValueError` while a sudo - `run_command(privileged=True, stdin=...)` raises `ValueError` while a sudo
+2 -2
View File
@@ -60,7 +60,7 @@ with Driver(
packages = dev.get_packages() packages = dev.get_packages()
updates = dev.get_pending_updates() updates = dev.get_pending_updates()
services = dev.get_services() services = dev.get_services()
users = dev.get_users() users = dev.get_os_users()
procs = dev.get_processes() procs = dev.get_processes()
jobs = dev.get_cron_jobs() jobs = dev.get_cron_jobs()
@@ -106,7 +106,7 @@ with Driver(
| `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman | | `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
| `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) | | `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) |
| `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable | | `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable |
| `get_users()` | ✅ | `/etc/passwd` + `/etc/group` | | `get_os_users()` | ✅ | `/etc/passwd` + `/etc/group` |
| `get_processes()` | ✅ | `ps axo` | | `get_processes()` | ✅ | `ps axo` |
| `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` | | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
+19 -3
View File
@@ -125,7 +125,14 @@ _REFRESH = {
"yum": "yum makecache -q 2>&1", "yum": "yum makecache -q 2>&1",
"apk": "apk update -q 2>&1", "apk": "apk update -q 2>&1",
} }
_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null" #: yum's counterpart of DNF_SECURITY_COMMAND, in the same shape: parse_dnf_security
#: (napalm-device-types 4.7) needs the ``__DNF_RC=`` line, and behind ``| cat`` it is
#: the only place yum's own exit status survives.
_YUM_SECURITY_COMMAND = (
"{ LC_ALL=C yum updateinfo list security -q 2>/dev/null; echo __DNF_RC=$?; } | cat"
)
#: The exit status ``DNF_SECURITY_COMMAND`` and ``_YUM_SECURITY_COMMAND`` report.
_DNF_RC_RE = re.compile(r"^__DNF_RC=(\d+)\s*$", re.MULTILINE)
#: Restart the host two seconds later, detached from this session: the launcher's #: Restart the host two seconds later, detached from this session: the launcher's
#: exit status comes back before the host goes down, and closing the session #: exit status comes back before the host goes down, and closing the session
@@ -1512,7 +1519,16 @@ class LinuxDriver(
"""Packages a pending security advisory covers; None when dnf/yum cannot say.""" """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 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)) output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
return parse_dnf_security(output) if status == 0 else None # Both commands pipe through cat, so the outer status is cat's: dnf/yum's own
# is in __DNF_RC. A failed run prints nothing on stdout and must not read as
# "no advisories"; neither may output cut short before the marker.
rc = _DNF_RC_RE.findall(output)
if status != 0 or not rc or rc[-1] != "0":
return None
try:
return parse_dnf_security(output)
except ValueError:
return None
def refresh_available_updates(self) -> dict[str, Any]: def refresh_available_updates(self) -> dict[str, Any]:
"""Refresh the package index (apt-get update, dnf makecache, apk update).""" """Refresh the package index (apt-get update, dnf makecache, apk update)."""
@@ -1695,7 +1711,7 @@ class LinuxDriver(
# OSDriver – users # OSDriver – users
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def get_users(self) -> list[UserDict]: def get_os_users(self) -> list[UserDict]:
"""Return local user accounts from /etc/passwd plus supplementary groups.""" """Return local user accounts from /etc/passwd plus supplementary groups."""
passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd") passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd")
groups_out = self._send("getent group 2>/dev/null || cat /etc/group") groups_out = self._send("getent group 2>/dev/null || cat /etc/group")
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "napalm-linux" name = "napalm-linux"
version = "0.5.0" version = "0.6.0"
description = "NAPALM driver for generic Linux systems via SSH" description = "NAPALM driver for generic Linux systems via SSH"
readme = "README.md" readme = "README.md"
requires-python = ">=3.9" requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
] ]
dependencies = [ dependencies = [
"napalm>=4.0", "napalm>=4.0",
"napalm-device-types>=4.5.0", "napalm-device-types>=4.7.1",
"netmiko>=4.0.0", "netmiko>=4.0.0",
"paramiko>=5.0.0", # CVE-2026-44405 "paramiko>=5.0.0", # CVE-2026-44405
] ]
+6
View File
@@ -42,9 +42,15 @@ class FakeChannel:
def recv_ready(self): def recv_ready(self):
return False return False
def recv(self, size):
return b""
def recv_stderr_ready(self): def recv_stderr_ready(self):
return False return False
def recv_stderr(self, size):
return b""
def exit_status_ready(self): def exit_status_ready(self):
return True return True
+88 -8
View File
@@ -4,7 +4,12 @@ import re
import pytest import pytest
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model, _split_status from napalm_linux.linux import (
_YUM_SECURITY_COMMAND,
LinuxDriver,
_arm_vendor_from_model,
_split_status,
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -93,6 +98,9 @@ class _AnsweringChannel:
def recv_stderr_ready(self): def recv_stderr_ready(self):
return False return False
def recv_stderr(self, size):
return b""
def exit_status_ready(self): def exit_status_ready(self):
return not self._out return not self._out
@@ -309,7 +317,7 @@ def test_get_available_updates_apt(driver):
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# get_users # get_os_users
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -326,9 +334,9 @@ GROUP_OUT = (
) )
def test_get_users(driver): def test_get_os_users(driver):
with patch.object(driver, "_send", side_effect=[PASSWD_OUT, GROUP_OUT]): with patch.object(driver, "_send", side_effect=[PASSWD_OUT, GROUP_OUT]):
users = driver.get_users() users = driver.get_os_users()
admin = next(u for u in users if u["username"] == "admin") admin = next(u for u in users if u["username"] == "admin")
assert admin["uid"] == 1000 assert admin["uid"] == 1000
@@ -338,6 +346,14 @@ def test_get_users(driver):
assert set(admin["groups"]) == {"sudo", "docker", "adm"} assert set(admin["groups"]) == {"sudo", "docker", "adm"}
def test_get_users_is_left_to_napalm(driver):
"""napalm-device-types 4.7 renamed OSDriver.get_users to get_os_users:
NAPALM's own get_users has another shape, and this driver does not implement it."""
assert LinuxDriver.get_users is not LinuxDriver.get_os_users
with pytest.raises(NotImplementedError):
driver.get_users()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# ping parsing # ping parsing
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -1460,9 +1476,12 @@ class TestAvailableUpdates:
driver._pkg_manager = "dnf" driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = [ driver._device.send_command.side_effect = [
"0", # id -u "0", # id -u
"openssl-libs.x86_64 1:3.1.4-2.fc40 updates\n" ("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", "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", # DNF_SECURITY_COMMAND pipes through cat: the outer status is cat's, dnf's own
# is in the __DNF_RC marker.
("FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
] ]
updates = {u["name"]: u for u in driver.get_available_updates()} updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1475,7 +1494,62 @@ class TestAvailableUpdates:
driver._device.send_command.side_effect = [ driver._device.send_command.side_effect = [
"0", "0",
"vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100", "vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
"Error: updateinfo metadata missing\n__NETORK_RC=1", "Error: updateinfo metadata missing\n__DNF_RC=1\n__NETORK_RC=0",
]
assert driver.get_available_updates()[0]["security"] is None
def test_yum_security_command_reports_its_own_exit_status(self):
# parse_dnf_security (napalm-device-types 4.7) needs the __DNF_RC marker, and
# behind "| cat" the outer status no longer says whether yum failed.
assert "echo __DNF_RC=$?" in _YUM_SECURITY_COMMAND
assert "yum updateinfo list security" in _YUM_SECURITY_COMMAND
def test_yum_marks_what_a_security_advisory_covers(self, driver):
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0", # id -u
("openssl-libs.x86_64 1:1.0.2k-26.el7_9 updates\n"
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100"),
("RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
]
updates = {u["name"]: u for u in driver.get_available_updates()}
assert driver._device.send_command.call_count == 3
assert "echo __DNF_RC=$?" in driver._device.send_command.call_args_list[2][0][0]
assert updates["openssl-libs"]["security"] is True
assert updates["vim-enhanced"]["security"] is False
def test_yum_with_no_advisories_marks_nothing_as_security(self, driver):
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0",
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100",
"__DNF_RC=0\n__NETORK_RC=0",
]
assert driver.get_available_updates()[0]["security"] is False
def test_yum_that_failed_leaves_security_unknown(self, driver):
# A failed yum prints nothing on stdout (2>/dev/null): read as "no advisories",
# every update would be marked as not a security fix.
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0",
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100",
"__DNF_RC=1\n__NETORK_RC=0",
]
assert driver.get_available_updates()[0]["security"] is None
def test_yum_without_its_exit_status_leaves_security_unknown(self, driver):
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0",
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100",
"RHSA-2023:3556 Important/Sec. vim-enhanced-2:7.4.629-8.el7_9.x86_64\n__NETORK_RC=0",
] ]
assert driver.get_available_updates()[0]["security"] is None assert driver.get_available_updates()[0]["security"] is None
@@ -1725,9 +1799,15 @@ class TestGetVirtualIps:
def recv_ready(self): def recv_ready(self):
return False return False
def recv(self, size):
return b""
def recv_stderr_ready(self): def recv_stderr_ready(self):
return False return False
def recv_stderr(self, size):
return b""
def exit_status_ready(self): def exit_status_ready(self):
return True return True
+3
View File
@@ -95,6 +95,9 @@ class _Channel:
def recv_stderr_ready(self): def recv_stderr_ready(self):
return False return False
def recv_stderr(self, size):
return b""
def exit_status_ready(self): def exit_status_ready(self):
return not self._out return not self._out
+3
View File
@@ -66,6 +66,9 @@ class Channel:
def recv_stderr_ready(self): def recv_stderr_ready(self):
return False return False
def recv_stderr(self, size):
return b""
def exit_status_ready(self): def exit_status_ready(self):
return not self._out return not self._out