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]
|
## [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
|
## [0.3.0] – 2026-10-08
|
||||||
|
|
||||||
### Removed
|
### Removed
|
||||||
|
|||||||
@@ -44,6 +44,7 @@ from napalm_device_types import (
|
|||||||
OSDriver,
|
OSDriver,
|
||||||
SystemdServicesMixin,
|
SystemdServicesMixin,
|
||||||
SystemdUnavailable,
|
SystemdUnavailable,
|
||||||
|
VirtualIpsMixin,
|
||||||
parse_apt_upgradable,
|
parse_apt_upgradable,
|
||||||
open_stream_on_transport,
|
open_stream_on_transport,
|
||||||
parse_dnf_security,
|
parse_dnf_security,
|
||||||
@@ -206,6 +207,7 @@ def _arm_vendor_from_model(model: str) -> str:
|
|||||||
class LinuxDriver(
|
class LinuxDriver(
|
||||||
KernelFactsMixin,
|
KernelFactsMixin,
|
||||||
ListeningSocketsMixin,
|
ListeningSocketsMixin,
|
||||||
|
VirtualIpsMixin,
|
||||||
SystemdServicesMixin,
|
SystemdServicesMixin,
|
||||||
HostStatusMixin,
|
HostStatusMixin,
|
||||||
ContainerEngineMixin,
|
ContainerEngineMixin,
|
||||||
@@ -1022,6 +1024,15 @@ class LinuxDriver(
|
|||||||
return self._run_privileged(command, 60)
|
return self._run_privileged(command, 60)
|
||||||
return self._send(command, read_timeout=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
|
# OSDriver – package management
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|||||||
+2
-2
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "napalm-linux"
|
name = "napalm-linux"
|
||||||
version = "0.3.0"
|
version = "0.4.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>=2.6.0",
|
"napalm-device-types>=4.2.0",
|
||||||
"netmiko>=4.0.0",
|
"netmiko>=4.0.0",
|
||||||
"paramiko>=5.0.0", # CVE-2026-44405
|
"paramiko>=5.0.0", # CVE-2026-44405
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -1575,3 +1575,101 @@ class TestGetListeningSockets:
|
|||||||
refused, plain = self._sent(driver)
|
refused, plain = self._sent(driver)
|
||||||
assert refused.startswith("sudo -n sh -c '")
|
assert refused.startswith("sudo -n sh -c '")
|
||||||
assert plain.startswith("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