feat: read the virtual IPs vip-manager and keepalived declare on a host

A virtual IP sits on whichever member holds it right now; a standby's address list shows no trace of it. What every member has is the configuration that declares it, and reading that is the same on every Linux host -- so VirtualIpsMixin.get_virtual_ips() lives here and a driver only carries the command across (NetOrk/netork#829). - vip-manager: every vip-manager.service / vip-manager@*.service unit; the address from a flag (1.x's -ip=${VIP_IP} resolved through the unit's environment), the environment, or the --config YAML (5.x). - keepalived: the virtual_ipaddress(_excluded) entries of every vrrp_instance, with its interface and virtual_router_id; include is followed four levels deep, and a pattern that is no plain path glob is never handed to the root shell. The same files hold the etcd password and auth_pass, so an awk program filters on the host and prints allowlisted fields only. None per mechanism is "did not look", [] a host without it. The command is about 4 kB and goes on an exec channel. Version 4.2.0.
2026-10-08 12:37:28 +02:00
parent 8111c4cde3
commit 37c09d0fa0
6 changed files with 1073 additions and 1 deletions
@@ -0,0 +1,617 @@
+"""get_virtual_ips: the addresses a host's HA configuration lets float between machines.
+
+A virtual IP is not visible from the address list alone: vip-manager and
+keepalived put it on whichever node is master *right now*, and a standby has no
+trace of it in ``ip addr``. What every member has is the configuration that
+declares it. Reading that is the same on every Linux host, so the command and
+its parse live here once, and a driver only carries the command across
+(netOrk #829).
+
+The configuration also holds secrets -- vip-manager's etcd password, keepalived's
+``auth_pass`` -- so the command filters on the host: only allowlisted fields
+ever leave it. The tests below run the command for real to hold it to that.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.virtual_ips import (
+    VIRTUAL_IPS_COMMAND,
+    VirtualIpsMixin,
+    parse_virtual_ips,
+)
+
+SECRET = "s3cr3t-do-not-leak"
+
+
+def _wire(*sections: str, noise: str = "") -> str:
+    """The report as the command prints it, framed."""
+    return f"{noise}VIPS_BEGIN\n{''.join(sections)}VIPS_END\n"
+
+
+def _section(mechanism: str, unit: str, *records: str) -> str:
+    return f"[{mechanism} {unit}]\n" + "".join(f"{r}\n" for r in records)
+
+
+def _vip_manager(*records: str, unit: str = "vip-manager.service") -> str:
+    return _section("vip-manager", unit, *records)
+
+
+def _keepalived(*records: str, unit: str = "keepalived.service") -> str:
+    return _section("keepalived", unit, *records)
+
+
+DB01 = _vip_manager(
+    "active active",
+    "arg config /etc/default/vip-manager.yml",
+    "cfg ip 10.7.224.10",
+    "cfg netmask 24",
+    "cfg interface ens7",
+    'cfg trigger-key "/service/netork-db/leader"',
+)
+
+
+class TestVipManager:
+    def test_the_5x_yaml_declares_the_address(self):
+        reading = parse_virtual_ips(_wire(DB01))
+
+        assert reading == {
+            "vip_manager": [
+                {
+                    "mechanism": "vip-manager",
+                    "instance": "vip-manager",
+                    "address": "10.7.224.10",
+                    "prefix_length": 24,
+                    "interface": "ens7",
+                    "group_key": "/service/netork-db/leader",
+                    "running": True,
+                }
+            ],
+            "keepalived": [],
+        }
+
+    def test_the_1x_unit_takes_its_values_from_the_environment_file(self):
+        """Ubuntu's 1.0.2 package: ``-ip=${VIP_IP}`` on the command line,
+        the values in /etc/default/vip-manager."""
+        section = _vip_manager(
+            "active active",
+            "arg ip ${VIP_IP}",
+            "arg mask $VIP_MASK",
+            "arg iface ${VIP_IFACE}",
+            "arg key ${VIP_KEY}",
+            "env VIP_IP 10.0.0.10",
+            "env VIP_MASK 255.255.255.0",
+            "env VIP_IFACE eth0",
+            'env VIP_KEY "/service/pg/leader"',
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.10",
+            24,
+            "eth0",
+        )
+        assert entry["group_key"] == "/service/pg/leader"
+
+    def test_a_flag_beats_the_environment_which_beats_the_config_file(self):
+        section = _vip_manager(
+            "active active",
+            "arg ip 10.0.0.1",
+            "env VIP_IP 10.0.0.2",
+            "env VIP_NETMASK 16",
+            "cfg ip 10.0.0.3",
+            "cfg netmask 24",
+            "cfg interface eth1",
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.1",
+            16,
+            "eth1",
+        )
+
+    @pytest.mark.parametrize(
+        "value, expected",
+        [
+            ("10.0.0.10 # the database", "10.0.0.10"),
+            ("'10.0.0.10'", "10.0.0.10"),
+            ('"fd00::10"', "fd00::10"),
+            ("FD00:0:0::10", "fd00::10"),
+        ],
+    )
+    def test_values_lose_quotes_and_comments(self, value, expected):
+        section = _vip_manager("active active", f"cfg ip {value}")
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["address"] == expected
+
+    def test_a_stopped_unit_still_declares_its_address(self):
+        """A standby whose vip-manager is down is still a member -- one that
+        cannot take the address over, which is worth knowing."""
+        section = _vip_manager("active inactive", "cfg ip 10.0.0.10")
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["running"] is False
+        assert entry["prefix_length"] is None
+
+    def test_a_unit_without_an_address_declares_nothing(self):
+        section = _vip_manager("active active", "arg ip ${VIP_IP}", "cfg netmask 24")
+
+        assert parse_virtual_ips(_wire(section))["vip_manager"] == []
+
+    def test_a_template_unit_is_named_after_its_instance(self):
+        section = _vip_manager(
+            "active active", "cfg ip 10.0.0.10", unit="vip-manager@pg16.service"
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["instance"] == "vip-manager@pg16"
+
+    def test_an_unreadable_config_is_not_looked_at(self):
+        """None, not []: the reading must not say the address is gone."""
+        section = _vip_manager("active active", "unreadable /etc/default/vip-manager.yml")
+
+        reading = parse_virtual_ips(_wire(section, _keepalived("active inactive")))
+
+        assert reading["vip_manager"] is None
+        assert reading["keepalived"] == []
+
+
+class TestKeepalived:
+    TWO_INSTANCES = _keepalived(
+        "active active",
+        "instance VI_DB",
+        "interface eth0",
+        "virtual_router_id 51",
+        "vip 10.0.0.10/24 dev eth0",
+        "vip 10.0.0.11",
+        "end",
+        "instance web",
+        "vip fd00::80",
+        "interface eth1",
+        "end",
+    )
+
+    def test_each_address_of_each_instance(self):
+        reading = parse_virtual_ips(_wire(self.TWO_INSTANCES))
+
+        assert reading["vip_manager"] == []
+        assert reading["keepalived"] == [
+            {
+                "mechanism": "keepalived",
+                "instance": "VI_DB",
+                "address": "10.0.0.10",
+                "prefix_length": 24,
+                "interface": "eth0",
+                "group_key": "vrid:51",
+                "running": True,
+            },
+            {
+                "mechanism": "keepalived",
+                "instance": "VI_DB",
+                "address": "10.0.0.11",
+                "prefix_length": 32,
+                "interface": "eth0",
+                "group_key": "vrid:51",
+                "running": True,
+            },
+            {
+                "mechanism": "keepalived",
+                "instance": "web",
+                "address": "fd00::80",
+                "prefix_length": 128,
+                "interface": "eth1",
+                "group_key": None,
+                "running": True,
+            },
+        ]
+
+    def test_dev_beats_the_instance_interface(self):
+        section = _keepalived(
+            "active active", "instance a", "interface eth0", "vip 10.0.0.10/24 dev eth9", "end"
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["keepalived"]
+
+        assert entry["interface"] == "eth9"
+
+    @pytest.mark.parametrize("line", ["vip $VIP", "vip @node1", "vip not-an-address"])
+    def test_what_is_no_address_is_skipped(self, line):
+        """keepalived's ``$VAR`` substitution and ``@host`` conditionals are
+        not resolved here; an entry that is no address is no entry."""
+        section = _keepalived("active active", "instance a", line, "vip 10.0.0.12", "end")
+
+        assert [e["address"] for e in parse_virtual_ips(_wire(section))["keepalived"]] == [
+            "10.0.0.12"
+        ]
+
+    def test_an_instance_cut_off_by_the_end_of_the_file_still_counts(self):
+        section = _keepalived("active active", "instance a", "vip 10.0.0.10")
+
+        assert len(parse_virtual_ips(_wire(section))["keepalived"]) == 1
+
+    def test_without_a_config_file_it_declares_nothing(self):
+        section = _keepalived("active inactive", "missing /etc/keepalived/keepalived.conf")
+
+        assert parse_virtual_ips(_wire(section))["keepalived"] == []
+
+    @pytest.mark.parametrize(
+        "record", ["unreadable /etc/keepalived/conf.d/x.conf", "too-deep /etc/keepalived/a.conf"]
+    )
+    def test_an_include_it_could_not_follow_is_not_looked_at(self, record):
+        section = _keepalived("active active", "instance a", "vip 10.0.0.10", "end", record)
+
+        assert parse_virtual_ips(_wire(section))["keepalived"] is None
+
+
+class TestTheReport:
+    def test_no_units_is_looked_and_found_nothing(self):
+        assert parse_virtual_ips(_wire("[systemd]\n")) == {"vip_manager": [], "keepalived": []}
+
+    def test_a_host_without_systemd_is_not_looked_at(self):
+        assert parse_virtual_ips(_wire("[no-systemd]\n")) == {
+            "vip_manager": None,
+            "keepalived": None,
+        }
+
+    def test_noise_before_the_report_is_ignored(self):
+        reading = parse_virtual_ips(_wire(DB01, noise="$ sh -c '...'\nWelcome to Ubuntu\n"))
+
+        assert reading["vip_manager"][0]["address"] == "10.7.224.10"
+
+    def test_a_report_cut_short_raises_without_quoting_it(self):
+        """The output may carry configuration; an error message must not."""
+        with pytest.raises(ValueError) as caught:
+            parse_virtual_ips(f"VIPS_BEGIN\n{DB01}")
+
+        assert "10.7.224.10" not in str(caught.value)
+
+    def test_entries_are_sorted(self):
+        reading = parse_virtual_ips(
+            _wire(
+                _vip_manager("active active", "cfg ip 10.0.0.20", unit="vip-manager@b.service"),
+                _vip_manager("active active", "cfg ip 10.0.0.3", unit="vip-manager@a.service"),
+            )
+        )
+
+        assert [e["address"] for e in reading["vip_manager"]] == ["10.0.0.3", "10.0.0.20"]
+
+
+# ---------------------------------------------------------------------------
+# The command itself, run against a stub systemctl and real files
+# ---------------------------------------------------------------------------
+
+DB01_YAML = f"""# managed by hand, see infrastructure/docs/DATABASE_CLUSTER.md
+ip: 10.7.224.10
+netmask: 24
+interface: ens7
+trigger-key: "/service/netork-db/leader"
+trigger-value: "db-01"
+dcs-type: etcd
+dcs-endpoints:
+  - http://10.7.234.12:2379
+etcd-user: patroni
+etcd-password: {SECRET}
+"""
+
+KEEPALIVED_CONF = f"""! Configuration File for keepalived
+global_defs {{
+   notification_email {{ ops@example.org }}
+   router_id LVS_{SECRET}
+}}
+
+vrrp_script chk_haproxy {{
+    script "/usr/local/bin/check {SECRET}"
+}}
+
+vrrp_instance VI_WEB {{
+    state MASTER
+    interface eth0
+    virtual_router_id 51
+    priority 101
+    authentication {{
+        auth_type PASS
+        auth_pass {SECRET}
+    }}
+    virtual_ipaddress {{
+        192.0.2.10/24 dev eth0 label eth0:1
+    }}
+    notify_master "/etc/keepalived/master.sh {SECRET}"
+}}
+
+include conf.d/*.conf
+"""
+
+INCLUDED_CONF = """vrrp_instance VI_DB { interface eth1
+  virtual_router_id 52
+  virtual_ipaddress { 192.0.2.20 }
+  virtual_ipaddress_excluded {
+    2001:db8::20/64
+  }
+}
+"""
+
+
+def _awks() -> list:
+    found = []
+    if shutil.which("mawk"):
+        found.append(pytest.param(["mawk"], id="mawk"))
+    if shutil.which("busybox"):
+        found.append(pytest.param(["busybox", "awk"], id="busybox"))
+    if shutil.which("gawk"):
+        found.append(pytest.param(["gawk"], id="gawk"))
+    return found or [pytest.param(None, marks=pytest.mark.skip(reason="no awk to run"))]
+
+
+class _Host:
+    """A host as far as the command can tell: units, their properties, files."""
+
+    def __init__(self, root: Path, awk: list) -> None:
+        self.root = root
+        self.bin = root / "bin"
+        self.bin.mkdir()
+        self.state = root / "systemd"
+        self.state.mkdir()
+        (self.state / "units").write_text("")
+        stub = self.bin / "systemctl"
+        stub.write_text(
+            "#!/bin/sh\n"
+            f"d={self.state}\n"
+            'case "$1" in\n'
+            f'  list-units) {shutil.which("cat")} "$d/units" ;;\n'
+            f'  show) for last; do :; done; {shutil.which("cat")} "$d/show-$last" ;;\n'
+            "esac\n"
+        )
+        stub.chmod(0o755)
+        wrapper = self.bin / "awk"
+        program = " ".join([shutil.which(awk[0]) or awk[0], *awk[1:]])
+        wrapper.write_text(f'#!/bin/sh\nexec {program} "$@"\n')
+        wrapper.chmod(0o755)
+
+    def unit(self, name: str, *properties: str) -> None:
+        with (self.state / "units").open("a") as units:
+            units.write(f"{name} loaded active running {name}\n")
+        (self.state / f"show-{name}").write_text("".join(f"{p}\n" for p in properties))
+
+    def file(self, relative: str, content: str) -> Path:
+        path = self.root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content)
+        return path
+
+    def run(self, *, systemctl: bool = True) -> str:
+        if not systemctl:
+            (self.bin / "systemctl").unlink()
+        return subprocess.run(
+            ["/bin/sh", "-c", VIRTUAL_IPS_COMMAND],
+            capture_output=True,
+            text=True,
+            env={"PATH": str(self.bin)},
+            timeout=30,
+        ).stdout
+
+
+def _exec_start(*argv: str) -> str:
+    return (
+        f"ExecStart={{ path={argv[0]} ; argv[]={' '.join(argv)} ; ignore_errors=no ; "
+        "start_time=[Mon 2026-10-05 06:50:00 UTC] ; stop_time=[n/a] ; pid=758 ; "
+        "code=(null) ; status=0/0 }"
+    )
+
+
+@pytest.fixture(params=_awks())
+def host(request, tmp_path) -> _Host:
+    return _Host(tmp_path, request.param)
+
+
+class TestTheCommandOnAHost:
+    def test_vip_manager_5x_as_on_db_01(self, host):
+        config = host.file("etc/default/vip-manager.yml", DB01_YAML)
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start("/usr/bin/vip-manager", f"--config={config}"),
+            f"Environment=VIP_ETCD_PASSWORD={SECRET} VIP_INTERFACE=ens7",
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "10.7.234.12" not in out  # the etcd endpoints stay on the host
+        assert parse_virtual_ips(out)["vip_manager"] == [
+            {
+                "mechanism": "vip-manager",
+                "instance": "vip-manager",
+                "address": "10.7.224.10",
+                "prefix_length": 24,
+                "interface": "ens7",
+                "group_key": "/service/netork-db/leader",
+                "running": True,
+            }
+        ]
+
+    def test_vip_manager_1x_with_its_environment_file(self, host):
+        envfile = host.file(
+            "etc/default/vip-manager",
+            f'VIP_IP="10.0.0.10"\nexport VIP_MASK=24\nVIP_IFACE=eth0\nVIP_KEY=/service/pg/leader\n'
+            f"VIP_ETCD_PASSWORD={SECRET}\n",
+        )
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start(
+                "/usr/bin/vip-manager",
+                "-ip=${VIP_IP}",
+                "-mask=${VIP_MASK}",
+                "-iface=${VIP_IFACE}",
+                "-key=${VIP_KEY}",
+                "-etcd_password",
+                SECRET,
+            ),
+            f"EnvironmentFiles={envfile} (ignore_errors=yes)",
+            f"EnvironmentFiles={host.root}/etc/default/missing (ignore_errors=yes)",
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        [entry] = parse_virtual_ips(out)["vip_manager"]
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.10",
+            24,
+            "eth0",
+        )
+
+    def test_keepalived_with_includes(self, host):
+        config = host.file("etc/keepalived/keepalived.conf", KEEPALIVED_CONF)
+        host.file("etc/keepalived/conf.d/db.conf", INCLUDED_CONF)
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", "--dont-fork", "-f", str(config)),
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "MASTER" not in out and "priority" not in out
+        reading = parse_virtual_ips(out)
+        assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
+                for e in reading["keepalived"]] == [
+            ("VI_WEB", "192.0.2.10", 24, "eth0", "vrid:51"),
+            ("VI_DB", "192.0.2.20", 32, "eth1", "vrid:52"),
+            ("VI_DB", "2001:db8::20", 64, "eth1", "vrid:52"),
+        ]
+        assert reading["vip_manager"] == []
+
+    def test_an_include_that_includes_itself_is_not_followed_forever(self, host):
+        config = host.file(
+            "etc/keepalived/keepalived.conf",
+            "vrrp_instance a {\n virtual_ipaddress {\n 192.0.2.30\n }\n}\n"
+            "include keepalived.conf\n",
+        )
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", f"--use-file={config}"),
+        )
+
+        assert parse_virtual_ips(host.run())["keepalived"] is None
+
+    def test_an_include_pattern_is_never_run_as_shell(self, host):
+        """The include line comes from a file and the command runs as root."""
+        marker = host.root / "PWNED"
+        config = host.file(
+            "etc/keepalived/keepalived.conf",
+            f"include /etc/x.conf;touch${{IFS}}{marker}\ninclude $(touch {marker})\n",
+        )
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", "-f", str(config)),
+        )
+
+        out = host.run()
+
+        assert not marker.exists()
+        assert parse_virtual_ips(out)["keepalived"] is None
+
+    @pytest.mark.skipif(os.geteuid() == 0, reason="root reads every file")
+    def test_a_config_it_may_not_read_is_reported_as_such(self, host):
+        config = host.file("etc/default/vip-manager.yml", DB01_YAML)
+        config.chmod(0o000)
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start("/usr/bin/vip-manager", "--config", str(config)),
+        )
+
+        out = host.run()
+
+        assert parse_virtual_ips(out)["vip_manager"] is None
+
+    def test_no_units_at_all(self, host):
+        assert parse_virtual_ips(host.run()) == {"vip_manager": [], "keepalived": []}
+
+    def test_a_host_without_systemd(self, host):
+        assert parse_virtual_ips(host.run(systemctl=False)) == {
+            "vip_manager": None,
+            "keepalived": None,
+        }
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        assert "VIPS_BEGIN" not in VIRTUAL_IPS_COMMAND
+        assert "VIPS_END" not in VIRTUAL_IPS_COMMAND
+
+    def test_it_writes_and_changes_nothing(self):
+        for verb in ("sudo", " > ", ">>", "kill", "rm ", "start", "stop", "restart", "reload"):
+            assert verb not in VIRTUAL_IPS_COMMAND
+
+    def test_it_is_posix_sh(self):
+        result = subprocess.run(
+            ["sh", "-n", "-c", VIRTUAL_IPS_COMMAND], capture_output=True, text=True
+        )
+        assert result.returncode == 0, result.stderr
+
+    def test_it_is_one_line(self):
+        """About 4 kB with its awk program -- past the line a terminal takes, so a
+        driver sends it on an exec channel. One argument there, one line."""
+        driver = _Driver(_wire())
+        driver.get_virtual_ips()
+
+        [(command, _privileged)] = driver.calls
+        assert "\n" not in command
+
+
+class _Driver(VirtualIpsMixin):
+    def __init__(self, privileged: str, unprivileged: str = "") -> None:
+        self.answers = {True: privileged, False: unprivileged}
+        self.calls: list = []
+
+    def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
+        self.calls.append((command, privileged))
+        return self.answers[privileged]
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire(DB01))
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"][0]["address"] == "10.7.224.10"
+        [(command, privileged)] = driver.calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
+
+    def test_without_root_it_reads_what_it_can(self):
+        """The config may be root's alone; what is readable is still worth having."""
+        driver = _Driver(
+            "sudo: a password is required\n",
+            _wire(_vip_manager("active active", "unreadable /etc/default/vip-manager.yml")),
+        )
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"] is None
+        assert [p for _, p in driver.calls] == [True, False]
+
+    def test_not_every_os_driver_has_it(self):
+        assert not issubclass(OSDriver, VirtualIpsMixin)
+        assert not hasattr(OSDriver, "get_virtual_ips")