README.md

@@ -96,6 +96,17 @@ without `ss` is read with `netstat -lntup` (OpenWrt's busybox, old net-tools); o
 cgroup names the procd service (`/services/<name>/<instance>`). A host with neither raises
 `ListeningSocketsUnavailable`.
 
+`VirtualIpsMixin` (`get_virtual_ips`) reads the addresses a host's HA configuration lets
+float between machines: vip-manager (5.x's `--config` YAML, 1.x's `-ip=${VIP_IP}` resolved
+through the unit's environment) and keepalived (`virtual_ipaddress` of every
+`vrrp_instance`, `include` followed four levels deep). It reports what is *declared*, not
+what the host holds right now -- that is in its interface addresses. The same files hold
+the etcd password and `auth_pass`, so an awk program filters **on the host** and prints
+allowlisted fields only; a secret never reaches the caller. Root first, then without it.
+A mechanism it could not read completely is `None` ("did not look"), a host without it
+`[]`. The command is about 4 kB, so a driver sends it on an exec channel
+(`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
+
 **Update readers raise when they cannot read.** `get_available_updates` returns an empty
 list only when nothing is pending; netOrk keeps "pending since" per package, and an empty
 list for "don't know" would reset it.

napalm_device_types/__init__.py

@@ -52,6 +52,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.services.ServiceControlMixin`
 * :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
+* :class:`~napalm_device_types.virtual_ips.VirtualIpsMixin`
 
 Introspection -- :func:`~napalm_device_types.roles.roles_of`,
 :func:`~napalm_device_types.roles.role_keys_of` and
@@ -113,6 +114,11 @@ from napalm_device_types.systemd import (
     parse_systemd_services,
 )
 from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.virtual_ips import (
+    VIRTUAL_IPS_COMMAND,
+    VirtualIpsMixin,
+    parse_virtual_ips,
+)
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
@@ -170,6 +176,9 @@ __all__ = [
     "SystemdUnavailable",
     "parse_systemd_services",
     "UpdateMixin",
+    "VIRTUAL_IPS_COMMAND",
+    "VirtualIpsMixin",
+    "parse_virtual_ips",
     "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",

napalm_device_types/models.py

@@ -366,6 +366,35 @@ class ListeningSocketsDict(TypedDict):
     sockets: List[ListeningSocketDict]
 
 
+class VirtualIpDict(TypedDict):
+    """An address the host's HA configuration lets float between machines.
+
+    Declared, not observed: whether the host holds the address right now is in
+    its interface addresses, not here. Only these fields leave the host -- never
+    a password, an etcd endpoint or a notify script.
+    """
+
+    mechanism: str  # "vip-manager" | "keepalived"
+    instance: str  # the vrrp_instance name, or the vip-manager unit without ".service"
+    address: str  # normalised: "10.7.224.10", "fd00::10"
+    prefix_length: Optional[int]  # None where the configuration gives none
+    interface: Optional[str]
+    group_key: Optional[str]  # vip-manager's trigger-key, or "vrid:<n>" for keepalived
+    running: Optional[bool]  # the unit is active
+
+
+class VirtualIpsDict(TypedDict):
+    """What ``VirtualIpsMixin.get_virtual_ips`` read, per mechanism.
+
+    A list is what was looked at -- ``[]`` a host without that mechanism. None is
+    "did not look": a configuration it could not read, an include it could not
+    follow, a host without systemd. A consumer keeps what it knew for a None.
+    """
+
+    vip_manager: Optional[List[VirtualIpDict]]
+    keepalived: Optional[List[VirtualIpDict]]
+
+
 class PortForwardDict(TypedDict):
     """A port the WAN side can reach, forwarded to a host inside.
 

napalm_device_types/virtual_ips.py

@@ -0,0 +1,406 @@
+# -*- coding: utf-8 -*-
+"""Virtual IPs: the addresses a host's HA configuration lets float between machines.
+
+vip-manager and keepalived put a service address on whichever member is master
+*right now*. The address list of a standby shows no trace of it; what every
+member has is the configuration that declares it. That is read the same way on
+every Linux host, so the command and its parse live here once, and a driver
+only carries the command across.
+
+**Declared, not held.** The reading says which addresses the configuration
+lets onto this host. Whether the host holds one at the moment is in its
+interface addresses, which the caller already has.
+
+**Filtered on the host.** The same files hold secrets -- vip-manager's etcd
+password, keepalived's ``auth_pass``, notify scripts with their arguments. The
+command therefore never prints a configuration: an awk program reads the unit's
+properties and files and prints allowlisted fields only. A secret cannot reach
+the caller's memory, its log or an exception text, because it never leaves the
+host.
+
+**What is read.**
+
+- *vip-manager*: every ``vip-manager.service`` and ``vip-manager@*.service`` unit
+  systemd knows. The address comes from the first of: a command-line flag
+  (``--ip``; 1.x's ``-ip=${VIP_IP}`` resolved through the unit's environment), the
+  environment (``VIP_IP``, an ``EnvironmentFile`` over ``Environment``), the
+  ``--config`` YAML (5.x's ``ip``). Mask, interface and trigger key alike.
+- *keepalived*: ``keepalived.service``, its ``-f``/``--use-file`` or
+  ``/etc/keepalived/keepalived.conf``. The ``virtual_ipaddress`` and
+  ``virtual_ipaddress_excluded`` entries of every ``vrrp_instance``, with the
+  instance's ``interface`` and ``virtual_router_id``. ``include`` is followed
+  four levels deep, a relative pattern from the including file's directory. A
+  pattern that is not a plain path glob is not followed: it would otherwise be
+  handed to a root shell.
+
+**None is "did not look".** A file that exists but cannot be read, an include
+that cannot be followed, a host without systemd: that mechanism is None, and the
+caller keeps what it knew. ``[]`` is a host that has none.
+
+**Root, and without it.** The configuration may be root's alone. The whole
+script goes to the host as one ``sh -c`` argument; when the privileged call
+brings no report back, it runs again without privilege and reads what it can.
+
+**An exec channel, not a terminal.** With its awk program the command is about
+4 kB, past the line an interactive shell reliably takes. A driver sends it the
+way it runs any command that needs no PTY -- napalm-linux's ``run_command``.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import re
+from dataclasses import dataclass, field
+from shlex import quote
+from typing import Callable, Dict, List, Optional, Set, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import VirtualIpDict, VirtualIpsDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "VIPS_BEGIN"
+_END = "VIPS_END"
+_NO_SYSTEMD = "no-systemd"
+#: A record that says a mechanism was not fully read.
+_BLIND = frozenset({"unreadable", "too-deep"})
+_KEEPALIVED_CONF = "/etc/keepalived/keepalived.conf"
+_ENV_KEYS = "VIP_(IP|MASK|NETMASK|IFACE|INTERFACE|KEY|TRIGGER_KEY)"
+_CFG_KEYS = "(ip|netmask|mask|interface|iface|trigger-key|key)"
+
+#: One line of awk, fed a unit's ``systemctl show`` output with ``-v m=<mechanism>``.
+#: It prints records, never a line of configuration: ``active <state>``,
+#: ``arg <flag> <value>``, ``env <VIP_*> <value>``, ``cfg <key> <value>``,
+#: ``instance <name>`` / ``interface`` / ``virtual_router_id`` / ``vip <addr> [dev <if>]``
+#: / ``end``, and ``unreadable``/``too-deep``/``missing <path>``. No ``>`` but in
+#: ``2>/dev/null``: comparisons are written the other way round. ``\047`` is a
+#: single quote, so the program itself needs none.
+_AWK = " ".join(
+    (
+        r'function q(p) { return "\047" p "\047" }',
+        r'function ok(f) { if (f == "" || f ~ /\047/) { print "unreadable " f; return 0 }'
+        r' if (system("test -r " q(f)) == 0) return 1;'
+        r' if (system("test -e " q(f)) == 0) print "unreadable " f; else print "missing " f;'
+        r" return 0 }",
+        r"function rf(f, c,   l, r, k) { if (!ok(f)) return;"
+        r" while (0 < (r = (getline l < f))) {"
+        rf' if (c == "cfg" && l ~ /^[ \t]*{_CFG_KEYS}[ \t]*:/) {{ sub(/^[ \t]*/, "", l); k = l;'
+        r' sub(/[ \t]*:.*/, "", k); sub(/^[^:]*:[ \t]*/, "", l); print "cfg " k " " l }'
+        rf' else if (c == "env" && l ~ /^[ \t]*(export[ \t]+)?{_ENV_KEYS}[ \t]*=/) {{'
+        r' sub(/^[ \t]*(export[ \t]+)?/, "", l); k = l; sub(/[ \t]*=.*/, "", k);'
+        r' sub(/^[^=]*=[ \t]*/, "", l); print "env " k " " l } }'
+        r' if (r < 0) print "unreadable " f; close(f) }',
+        r'function push(n, a) { st[++sd] = n; if (n == "vrrp_instance") print "instance " a }',
+        r'function pop() { if (0 < sd) { if (st[sd] == "vrrp_instance") print "end"; sd-- } }',
+        r"function stmt(w, nw, f, d,   x, i) { if (!nw) return;"
+        r' if (w[1] == "include") { if (1 < nw) inc(w[2], f, d); return }'
+        r' if (sd == 1 && st[1] == "vrrp_instance") {'
+        r' if (1 < nw && (w[1] == "interface" || w[1] == "virtual_router_id")) print w[1] " " w[2];'
+        r" return }"
+        r' if (sd == 2 && st[1] == "vrrp_instance" && st[2] ~ /^virtual_ipaddress(_excluded)?$/) {'
+        r' x = "vip " w[1]; for (i = 2; i < nw; i++) if (w[i] == "dev") x = x " dev " w[i + 1];'
+        r" print x } }",
+        r"function rk(f, d,   l, r, n, t, k, w, nw) {"
+        r' if (4 < d || (f in seen)) { print "too-deep " f; return }'
+        r" if (!ok(f)) return; seen[f] = 1;"
+        r' while (0 < (r = (getline l < f))) { sub(/[#!].*/, "", l); gsub(/[{}]/, " & ", l);'
+        r' n = split(l, t, " "); nw = 0; for (k = 1; k <= n; k++) {'
+        r' if (t[k] == "{") { push(nw ? w[1] : "", 1 < nw ? w[2] : ""); nw = 0 }'
+        r' else if (t[k] == "}") { stmt(w, nw, f, d); nw = 0; pop() } else w[++nw] = t[k] }'
+        r" stmt(w, nw, f, d) }"
+        r' if (r < 0) print "unreadable " f; close(f); delete seen[f] }',
+        r"function inc(p, f, d,   c, g, dir, fs, n, i) {"
+        r' if (p !~ /^\//) { dir = f; sub(/[^\/]*$/, "", dir); p = dir p }'
+        r' if (p !~ /^[A-Za-z0-9_.\/*?-]+$/) { print "unreadable " p; return }'
+        r' c = "for g in " p "; do [ -e \"$g\" ] && echo \"$g\"; done"; n = 0;'
+        r" while (0 < (c | getline g)) fs[++n] = g; close(c);"
+        r" for (i = 1; i <= n; i++) rk(fs[i], d + 1) }",
+        r'/^ActiveState=/ { print "active " substr($0, 13); next }',
+        r'/^ExecStart=/ { s = $0; i = index(s, "argv[]="); if (!i) next; s = substr(s, i + 7);'
+        r' j = index(s, " ;"); if (j) s = substr(s, 1, j - 1); n = split(s, t, " ");'
+        r' for (k = 2; k <= n; k++) { f = t[k]; if (f !~ /^-/) continue; v = ""; x = index(f, "=");'
+        r" if (x) { v = substr(f, x + 1); f = substr(f, 1, x - 1) }"
+        r' else if (k < n && t[k + 1] !~ /^-/) v = t[++k]; sub(/^--?/, "", f);'
+        r" if (f ~ /^(config|ip|netmask|mask|interface|iface|trigger-key|key|f|use-file)$/) {"
+        r' print "arg " f " " v; a[f] = v } } next }',
+        r'/^EnvironmentFiles=/ { s = substr($0, 18); sub(/ \(ignore_errors=[a-z]*\)$/, "", s);'
+        r' if (s != "") ef[++ne] = s; next }',
+        r'/^Environment=/ { n = split(substr($0, 13), t, " "); for (k = 1; k <= n; k++) {'
+        rf' x = index(t[k], "="); if (x && substr(t[k], 1, x - 1) ~ /^{_ENV_KEYS}$/)'
+        r' print "env " substr(t[k], 1, x - 1) " " substr(t[k], x + 1) } next }',
+        rf'END {{ if (m == "keepalived") {{ p = "{_KEEPALIVED_CONF}";'
+        r' if ("use-file" in a) p = a["use-file"]; if ("f" in a) p = a["f"]; rk(p, 0) }'
+        r' else { for (k = 1; k <= ne; k++) rf(ef[k], "env");'
+        r' if ("config" in a) rf(a["config"], "cfg") } }',
+    )
+)
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two halves
+#: so that a transport which echoes the command does not show them early.
+VIRTUAL_IPS_COMMAND = (
+    "PATH=$PATH:/usr/sbin:/sbin; "
+    "printf '%s%s\\n' VIPS_ BEGIN; "
+    "if command -v systemctl >/dev/null 2>&1; then echo '[systemd]'; "
+    "for u in $(systemctl list-units --all --plain --no-legend --type=service "
+    "vip-manager.service 'vip-manager@*' keepalived.service 2>/dev/null | awk '{print $1}'); do "
+    "case $u in keepalived*) m=keepalived;; *) m=vip-manager;; esac; "
+    'echo "[$m $u]"; '
+    'systemctl show -p ActiveState -p ExecStart -p EnvironmentFiles -p Environment "$u" '
+    f"2>/dev/null | awk -v m=$m {quote(_AWK)}; done; "
+    "else echo '[no-systemd]'; fi; "
+    "printf '%s%s\\n' VIPS_ END"
+)
+
+_COMMENT_RE = re.compile(r"\s+#.*$")
+_VARIABLE_RE = re.compile(r"^\$\{?(\w+)\}?$")
+_Record = Tuple[str, str]
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact virtual IP report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> List[Tuple[str, str, List[_Record]]]:
+    """``[<mechanism> <unit>]`` headers, each with its records as ``(kind, rest)``."""
+    sections: List[Tuple[str, str, List[_Record]]] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            mechanism, _, unit = line[1:-1].partition(" ")
+            sections.append((mechanism, unit, []))
+        elif line and sections:
+            kind, _, rest = line.partition(" ")
+            sections[-1][2].append((kind, rest.strip()))
+    return sections
+
+
+def _clean(value: str) -> str:
+    """Without a trailing ``# comment`` and the quotes around it."""
+    value = _COMMENT_RE.sub("", value.strip())
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
+        value = value[1:-1]
+    return value.strip()
+
+
+def _address(text: str) -> Optional[Tuple[str, Optional[int]]]:
+    """``"10.0.0.10/24"`` -> ``("10.0.0.10", 24)``; None for what is no address."""
+    host, slash, prefix = text.partition("/")
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        return None
+    if not slash:
+        return str(address), None
+    if not prefix.isdigit() or int(prefix) > address.max_prefixlen:
+        return None
+    return str(address), int(prefix)
+
+
+def _prefix(mask: Optional[str]) -> Optional[int]:
+    """``"24"`` or ``"255.255.255.0"`` -> 24."""
+    if not mask:
+        return None
+    if mask.isdigit():
+        return int(mask) if int(mask) <= 128 else None
+    try:
+        return ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen
+    except ValueError:
+        return None
+
+
+def _running(records: List[_Record]) -> Optional[bool]:
+    states = [rest for kind, rest in records if kind == "active"]
+    return states[-1] == "active" if states else None
+
+
+def _vip_manager(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
+    values: Dict[str, Dict[str, str]] = {"arg": {}, "env": {}, "cfg": {}}
+    for kind, rest in records:
+        if kind in values:
+            key, _, value = rest.partition(" ")
+            values[kind][key] = value
+    env = values["env"]
+
+    def resolve(value: str) -> Optional[str]:
+        value = _clean(value)
+        variable = _VARIABLE_RE.match(value)
+        if variable:
+            value = _clean(env.get(variable.group(1), ""))
+        return value if value and "$" not in value else None
+
+    def pick(args: Tuple[str, ...], envs: Tuple[str, ...], cfgs: Tuple[str, ...]) -> Optional[str]:
+        for kind, names in (("arg", args), ("env", envs), ("cfg", cfgs)):
+            for name in names:
+                if name in values[kind]:
+                    found = resolve(values[kind][name])
+                    if found:
+                        return found
+        return None
+
+    ip = pick(("ip",), ("VIP_IP",), ("ip",))
+    parsed = _address(ip) if ip else None
+    if parsed is None:
+        return []
+    address, prefix = parsed
+    mask = pick(("netmask", "mask"), ("VIP_NETMASK", "VIP_MASK"), ("netmask", "mask"))
+    return [
+        {
+            "mechanism": "vip-manager",
+            "instance": unit[: -len(".service")] if unit.endswith(".service") else unit,
+            "address": address,
+            "prefix_length": _prefix(mask) if mask else prefix,
+            "interface": pick(
+                ("interface", "iface"), ("VIP_INTERFACE", "VIP_IFACE"), ("interface", "iface")
+            ),
+            "group_key": pick(
+                ("trigger-key", "key"), ("VIP_TRIGGER_KEY", "VIP_KEY"), ("trigger-key", "key")
+            ),
+            "running": _running(records),
+        }
+    ]
+
+
+@dataclass
+class _Instance:
+    name: str
+    interface: Optional[str] = None
+    vrid: str = ""
+    vips: List[str] = field(default_factory=list)
+
+
+def _keepalived(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
+    running = _running(records)
+    instances: List[_Instance] = []
+    current: Optional[_Instance] = None
+    for kind, rest in records:
+        if kind == "instance":
+            current = _Instance(rest or unit)
+            instances.append(current)
+        elif current is None:
+            continue
+        elif kind == "interface":
+            current.interface = rest or None
+        elif kind == "virtual_router_id":
+            current.vrid = rest
+        elif kind == "vip":
+            current.vips.append(rest)
+        elif kind == "end":
+            current = None
+    return [entry for instance in instances for entry in _instance_entries(instance, running)]
+
+
+def _instance_entries(instance: _Instance, running: Optional[bool]) -> List[VirtualIpDict]:
+    entries: List[VirtualIpDict] = []
+    for vip in instance.vips:
+        tokens = vip.split()
+        parsed = _address(tokens[0]) if tokens else None
+        if parsed is None:
+            continue
+        address, prefix = parsed
+        dev = tokens[tokens.index("dev") + 1] if "dev" in tokens[1:-1] else None
+        entries.append(
+            {
+                "mechanism": "keepalived",
+                "instance": instance.name,
+                "address": address,
+                "prefix_length": prefix if prefix is not None else _host_prefix(address),
+                "interface": dev or instance.interface,
+                "group_key": f"vrid:{instance.vrid}" if instance.vrid.isdigit() else None,
+                "running": running,
+            }
+        )
+    return entries
+
+
+def _host_prefix(address: str) -> int:
+    """keepalived's default for an entry without one: the single address."""
+    return ipaddress.ip_address(address).max_prefixlen
+
+
+_READERS: Dict[str, Tuple[str, Callable[[str, List[_Record]], List[VirtualIpDict]]]] = {
+    "vip-manager": ("vip_manager", _vip_manager),
+    "keepalived": ("keepalived", _keepalived),
+}
+
+
+def _order(entry: VirtualIpDict) -> Tuple[int, int, str]:
+    address = ipaddress.ip_address(entry["address"])
+    return address.version, int(address), entry["instance"]
+
+
+def parse_virtual_ips(output: str) -> VirtualIpsDict:
+    """Parse what :data:`VIRTUAL_IPS_COMMAND` printed, each mechanism's entries
+    sorted by address.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    if any(mechanism == _NO_SYSTEMD for mechanism, _, _ in sections):
+        return {"vip_manager": None, "keepalived": None}
+    found: Dict[str, List[VirtualIpDict]] = {"vip_manager": [], "keepalived": []}
+    blind: Set[str] = set()
+    for mechanism, unit, records in sections:
+        if mechanism not in _READERS:
+            continue
+        key, read = _READERS[mechanism]
+        if any(kind in _BLIND for kind, _ in records):
+            blind.add(key)
+        else:
+            found[key].extend(read(unit, records))
+    def result(key: str) -> Optional[List[VirtualIpDict]]:
+        return None if key in blind else sorted(found[key], key=_order)
+
+    return {"vip_manager": result("vip_manager"), "keepalived": result("keepalived")}
+
+
+class VirtualIpsMixin:
+    """Adds :meth:`get_virtual_ips` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the command and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_virtual_ips_command` -- how a command reaches its host, and how it
+    gains root there. Mixed in by the drivers that can, not by
+    :class:`~napalm_device_types.os.OSDriver`, so ``hasattr(driver,
+    "get_virtual_ips")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root
+            when *privileged*. The command is a single ``sh -c`` invocation of
+            about 4 kB: send it on an exec channel, not typed into a shell."""
+            ...
+
+    def get_virtual_ips(self) -> VirtualIpsDict:
+        """
+        Returns the virtual IPs the host's vip-manager and keepalived configuration
+        declares -- whether or not the host holds them at the moment.
+
+        * vip_manager (list or None) - see :class:`~napalm_device_types.models.VirtualIpDict`
+        * keepalived (list or None)
+
+        None for a mechanism means it could not be read; ``[]`` that the host has none.
+
+        Example::
+
+            {
+                "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": [],
+            }
+
+        :raises ValueError: if neither reading carried an intact report.
+        """
+        command = f"sh -c {quote(VIRTUAL_IPS_COMMAND)}"
+        try:
+            return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=True))
+        except ValueError:
+            pass
+        return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=False))

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.1.0"
+version = "4.2.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_virtual_ips.py

@@ -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")