From a78daccb56f60b3b36257d87cc13f1627d1ff3a6 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 8 Oct 2026 12:38:03 +0200 Subject: [PATCH] feat: report the virtual IPs vip-manager and keepalived declare LinuxDriver mixes in napalm-device-types' VirtualIpsMixin (4.2.0) and carries its command on the exec channel: with the awk filter it is about 4 kB, past the line the PTY netmiko holds reliably takes. Root first, then without it (NetOrk/netork#829). Version 0.4.0. --- CHANGELOG.md | 10 +++++ napalm_linux/linux.py | 11 +++++ pyproject.toml | 4 +- tests/test_linux.py | 98 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 121 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a64bae5..b7833bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.4.0] – 2026-10-08 + +### Added + +- `get_virtual_ips()` through napalm-device-types' `VirtualIpsMixin` (4.2.0): the + addresses vip-manager and keepalived configuration lets float onto this host, + whether or not it holds them right now (NetOrk/netork#829). The command goes on + the exec channel (`run_command`): with its awk filter it is about 4 kB, past the + line the PTY reliably takes. Root first, then without it. + ## [0.3.0] – 2026-10-08 ### Removed diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 27cef79..859e61e 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -44,6 +44,7 @@ from napalm_device_types import ( OSDriver, SystemdServicesMixin, SystemdUnavailable, + VirtualIpsMixin, parse_apt_upgradable, open_stream_on_transport, parse_dnf_security, @@ -206,6 +207,7 @@ def _arm_vendor_from_model(model: str) -> str: class LinuxDriver( KernelFactsMixin, ListeningSocketsMixin, + VirtualIpsMixin, SystemdServicesMixin, HostStatusMixin, ContainerEngineMixin, @@ -1022,6 +1024,15 @@ class LinuxDriver( return self._run_privileged(command, 60) return self._send(command, read_timeout=60) + def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str: + """The transport for ``VirtualIpsMixin.get_virtual_ips``: an exec channel. + + The command carries its awk program and is about 4 kB, past the line the + PTY netmiko holds reliably takes. Read-only either way; root only because + the configuration may be root's alone. + """ + return self.run_command(command, privileged=privileged, timeout=60).stdout + # ------------------------------------------------------------------ # OSDriver – package management # ------------------------------------------------------------------ diff --git a/pyproject.toml b/pyproject.toml index 730db4e..c1a878b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-linux" -version = "0.3.0" +version = "0.4.0" description = "NAPALM driver for generic Linux systems via SSH" readme = "README.md" requires-python = ">=3.9" @@ -37,7 +37,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0", - "napalm-device-types>=2.6.0", + "napalm-device-types>=4.2.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 ] diff --git a/tests/test_linux.py b/tests/test_linux.py index f0997c8..2122c1b 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -1575,3 +1575,101 @@ class TestGetListeningSockets: refused, plain = self._sent(driver) assert refused.startswith("sudo -n sh -c '") assert plain.startswith("sh -c '") + + +# --------------------------------------------------------------------------- +# get_virtual_ips (napalm-device-types VirtualIpsMixin) +# --------------------------------------------------------------------------- + +_VIPS = ( + "VIPS_BEGIN\n[systemd]\n[vip-manager vip-manager.service]\nactive active\n" + "cfg ip 10.7.224.10\ncfg netmask 24\ncfg interface ens7\nVIPS_END\n" +) + + +class TestGetVirtualIps: + """The command is about 4 kB -- past the line a terminal takes -- so it goes + on an exec channel, never typed into the PTY netmiko holds.""" + + @staticmethod + def _exec(driver, *answers): + from napalm_device_types import CommandResult + + calls = [] + replies = iter(answers) + + def run_command(command, *, privileged=False, timeout=60, stdin=None): + calls.append((command, privileged)) + return CommandResult(*next(replies)) + + driver.run_command = run_command + return calls + + def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver): + from napalm_device_types import VirtualIpsMixin + + assert isinstance(driver, VirtualIpsMixin) + calls = self._exec(driver, (_VIPS, "", 0)) + + reading = driver.get_virtual_ips() + + assert reading["vip_manager"][0]["address"] == "10.7.224.10" + [(command, privileged)] = calls + assert privileged is True + assert command.startswith("sh -c '") + driver._device.send_command.assert_not_called() + + def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver): + calls = self._exec(driver, ("", "sudo: a password is required\n", 1), (_VIPS, "", 0)) + + reading = driver.get_virtual_ips() + + assert reading["vip_manager"][0]["interface"] == "ens7" + assert [privileged for _, privileged in calls] == [True, False] + + def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self): + from types import SimpleNamespace + + channels = [] + + class Channel: + def __init__(self): + channels.append(self) + + def exec_command(self, command): + self.command = command + + def settimeout(self, timeout): + pass + + def shutdown_write(self): + pass + + def close(self): + pass + + def recv_ready(self): + return False + + def recv_stderr_ready(self): + return False + + def exit_status_ready(self): + return True + + def recv_exit_status(self): + return 1 + + transport = SimpleNamespace(open_session=Channel) + driver = LinuxDriver("h", "u", "p", optional_args={}) + driver._device = SimpleNamespace( + remote_conn_pre=SimpleNamespace(get_transport=lambda: transport) + ) + driver._root = False + + with pytest.raises(ValueError): + driver.get_virtual_ips() + + privileged, plain = (c.command for c in channels) + assert privileged.startswith("sudo -n sh -c 'sh -c '") + assert plain.startswith("sh -c '")