feat: report the virtual IPs vip-manager and keepalived declare #22
@@ -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
|
||||
|
||||
@@ -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
@@ -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
|
||||
]
|
||||
|
||||
@@ -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 '")
|
||||
|
||||
Reference in New Issue
Block a user