feat: report the virtual IPs vip-manager and keepalived declare #22

Merged
christianmanivong merged 1 commits from feature/virtual-ips into master 2026-10-08 10:47:08 +00:00
4 changed files with 121 additions and 2 deletions
+10
View File
@@ -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
+11
View File
@@ -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
# ------------------------------------------------------------------
+2 -2
View File
@@ -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
]
+98
View File
@@ -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 '")