feat: report the firewall's CARP addresses as virtual IPs

get_virtual_ips() reads every address bound to a CARP VHID -- the CARP
address and IP aliases on its VHID alike -- from
/api/interfaces/overview/export, with this node's CARP state, so the reading
says who holds each address. interfaces/vip_settings would list the same
addresses with each VHID's CARP password; it is not read.

vip_manager and keepalived are Linux mechanisms and are None here. Needs
napalm-device-types 4.10.0 (carp_virtual_ip).

netOrk #1121.

README.md

@@ -89,6 +89,7 @@ arguments.
 | `get_mac_address_table` | ❌ | Not applicable (firewall, no L2 switching) |
 | `ping` | ✅ | `POST /api/diagnostics/ping/set` + `start` + `search_jobs` + `stop`/`remove` |
 | `get_listening_sockets` | ✅ | `GET /api/diagnostics/interface/get_socket_statistics` + `/api/core/service/search` + `/api/wireguard/service/show` (WireGuard's kernel sockets by listen port) |
+| `get_virtual_ips` | ✅ | `GET /api/interfaces/overview/export`: every address bound to a CARP VHID (CARP address and IP aliases on it), with this node's CARP state. `interfaces/vip_settings` is not read: it carries the CARP passwords |
 | `ping_sweep` | ✅ ³ | same endpoints, one batch of parallel jobs at a time |
 
 > ¹ Requires the `os-lldpd` plugin. Returns empty dict if the plugin is not installed.

napalm_opnsense/opnsense.py

@@ -53,12 +53,13 @@ import requests
 from requests.exceptions import RequestException
 
 from napalm_device_types import FingerprintRule, FirewallDriver
-from napalm_device_types.models import ListeningSocketsDict
+from napalm_device_types.models import ListeningSocketsDict, VirtualIpsDict
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException
 
 from napalm_opnsense.listening import parse_socket_statistics, wireguard_listen_ports
 from napalm_opnsense.ping_mixin import OPNsensePingMixin
 from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
+from napalm_opnsense.virtual_ips import carp_addresses
 
 
 class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
@@ -1824,6 +1825,22 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
             "sockets": parse_socket_statistics(statistics, services, wireguard),
         }
 
+    def get_virtual_ips(self) -> VirtualIpsDict:
+        """The firewall's CARP addresses, and whether this node holds each.
+
+        From ``/api/interfaces/overview/export``: every address bound to a CARP
+        VHID, the CARP address and IP aliases on its VHID alike, with this node's
+        CARP state (:mod:`napalm_opnsense.virtual_ips`). vip-manager and keepalived
+        are Linux mechanisms and are not looked for (None).
+
+        :raises: whatever the request raises -- an unread overview is not "no CARP".
+        """
+        return {
+            "vip_manager": None,
+            "keepalived": None,
+            "carp": carp_addresses(self._get("/api/interfaces/overview/export")),
+        }
+
     def manage_service(self, name: str, action: str) -> dict[str, Any]:
         """Execute a lifecycle action on an OPNsense service.
 

napalm_opnsense/virtual_ips.py

@@ -0,0 +1,63 @@
+"""The firewall's CARP addresses, read from its interface overview (netOrk #1121).
+
+``/api/interfaces/overview/export`` lists every address of an interface; one bound
+to a CARP VHID carries ``vhid`` and this node's CARP ``status``. That covers the
+CARP address itself and an IP alias bound to its VHID alike -- both float with the
+VHID's MASTER. An address without a VHID stays on this node and is no virtual IP.
+
+``interfaces/vip_settings/search_item`` would list the same addresses, but with
+each VHID's CARP password in it. The overview has none, so a secret never reaches
+the caller.
+
+What a state means (MASTER holds, BACKUP stands by) is the same on every CARP
+firewall: :func:`napalm_device_types.virtual_ips.carp_virtual_ip`.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+from typing import Any
+
+from napalm_device_types.models import VirtualIpDict
+from napalm_device_types.virtual_ips import carp_virtual_ip
+
+
+def _interfaces(overview: Any) -> list[Any]:
+    """The interface list, whether the answer is the list or wraps it."""
+    if isinstance(overview, dict):
+        return overview.get("interfaces") or []
+    if isinstance(overview, list):
+        return overview
+    raise ValueError(f"unexpected interface overview: {type(overview).__name__}")
+
+
+def carp_addresses(overview: Any) -> list[VirtualIpDict]:
+    """Every address bound to a CARP VHID in *overview*, sorted by address.
+
+    :raises ValueError: when the answer is neither the interface list nor wraps it.
+    """
+    entries: list[VirtualIpDict] = []
+    for interface in _interfaces(overview):
+        if not isinstance(interface, dict):
+            continue
+        identifier = interface.get("identifier") or ""
+        for family in ("ipv4", "ipv6"):
+            for address in interface.get(family) or []:
+                if not isinstance(address, dict) or not address.get("vhid"):
+                    continue
+                entry = carp_virtual_ip(
+                    address=str(address.get("ipaddr") or ""),
+                    prefix_length=None,
+                    vhid=address.get("vhid"),
+                    state=address.get("status"),
+                    instance=interface.get("description") or identifier,
+                    interface=interface.get("device") or identifier or None,
+                )
+                if entry is not None:
+                    entries.append(entry)
+    return sorted(entries, key=_order)
+
+
+def _order(entry: VirtualIpDict) -> tuple[int, int]:
+    address = ipaddress.ip_address(entry["address"])
+    return address.version, int(address)

pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=0.2.0",
+    "napalm_device_types>=4.10.0",
     "requests>=2.28.0",
 ]
 

tests/unit/fixtures/interfaces_overview_carp.json

@@ -0,0 +1,62 @@
+[
+  {
+    "identifier": "lan",
+    "device": "em0",
+    "description": "MGMT",
+    "addr4": "192.0.2.2/24",
+    "ipv4": [
+      {"ipaddr": "192.0.2.2/24"},
+      {"ipaddr": "192.0.2.1/24", "vhid": "108", "status": "MASTER", "advbase": "1", "advskew": "0", "peer": "224.0.0.18", "peer6": "ff02::12"}
+    ],
+    "ipv6": [
+      {"ipaddr": "fe80::1:2/64"},
+      {"ipaddr": "2001:db8:8::1/64", "vhid": "108", "status": "MASTER", "advbase": "1", "advskew": "0", "peer": "224.0.0.18", "peer6": "ff02::12"}
+    ]
+  },
+  {
+    "identifier": "opt2",
+    "device": "vlan0040",
+    "description": "SERVER",
+    "addr4": "198.51.100.2/24",
+    "ipv4": [
+      {"ipaddr": "198.51.100.2/24"},
+      {"ipaddr": "198.51.100.140/32", "vhid": "140", "status": "MASTER", "advbase": "1", "advskew": "0", "peer": "224.0.0.18", "peer6": "ff02::12"},
+      {"ipaddr": "198.51.100.1/24", "vhid": "140", "status": "MASTER", "advbase": "1", "advskew": "0", "peer": "224.0.0.18", "peer6": "ff02::12"}
+    ]
+  },
+  {
+    "identifier": "opt9",
+    "device": "vlan0011",
+    "description": "",
+    "addr4": "203.0.113.2/24",
+    "ipv4": [
+      {"ipaddr": "203.0.113.2/24"},
+      {"ipaddr": "203.0.113.1/24", "vhid": "111", "status": "INIT", "advbase": "1", "advskew": "0", "peer": "224.0.0.18", "peer6": "ff02::12"}
+    ]
+  },
+  {
+    "identifier": "opt3",
+    "device": "vlan0020",
+    "description": "IOT",
+    "addr4": "192.0.2.130/25",
+    "ipv4": [
+      {"ipaddr": "192.0.2.130/25"},
+      {"ipaddr": "192.0.2.140/32"}
+    ]
+  },
+  {
+    "identifier": "lo0",
+    "device": "lo0",
+    "description": "Loopback",
+    "addr4": "127.0.0.1/8",
+    "ipv4": [{"ipaddr": "127.0.0.1/8"}],
+    "ipv6": [{"ipaddr": "::1/128"}, {"ipaddr": "fe80::1/64"}]
+  },
+  {
+    "identifier": "wg1",
+    "device": "wg1",
+    "description": "WireGuard",
+    "addr4": "",
+    "ipv4": null
+  }
+]

tests/unit/test_virtual_ips.py

@@ -0,0 +1,130 @@
+"""The firewall's CARP addresses, with who holds each (netOrk #1121).
+
+``/api/interfaces/overview/export`` -- the source ``get_interfaces_ip`` reads --
+lists every address on an interface, and an address bound to a CARP VHID carries
+the VHID and this node's CARP state. That covers both the CARP address itself and
+an IP alias bound to its VHID: both float with the VHID's MASTER. An address
+without a VHID stays on this node and is no virtual IP.
+
+The overview has no CARP password in it; ``interfaces/vip_settings`` would, which
+is why it is not read. The fixture is the shape of a real OPNsense 26.7 answer,
+cut down, with documentation addresses.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+from napalm_opnsense.virtual_ips import carp_addresses
+
+OVERVIEW = json.loads(
+    (Path(__file__).parent / "fixtures" / "interfaces_overview_carp.json").read_text()
+)
+
+
+def _driver(overview=OVERVIEW) -> OPNsenseDriver:
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(hostname="fw", username="k", password="s")
+    drv.session = MagicMock()
+
+    def get(path: str):
+        if path == "/api/interfaces/overview/export":
+            if isinstance(overview, Exception):
+                raise overview
+            return overview
+        raise AssertionError(f"unexpected GET {path}")
+
+    drv._get = get  # type: ignore[method-assign]
+    return drv
+
+
+def _by_address(entries) -> dict:
+    return {e["address"]: e for e in entries}
+
+
+class TestCarpAddresses:
+    def test_every_address_bound_to_a_vhid_is_a_virtual_ip(self):
+        assert [e["address"] for e in carp_addresses(OVERVIEW)] == [
+            "192.0.2.1",
+            "198.51.100.1",
+            "198.51.100.140",
+            "203.0.113.1",
+            "2001:db8:8::1",
+        ]
+
+    def test_an_address_without_a_vhid_stays_on_this_node(self):
+        addresses = {e["address"] for e in carp_addresses(OVERVIEW)}
+        assert not addresses & {"192.0.2.2", "192.0.2.140", "127.0.0.1", "fe80::1"}
+
+    def test_the_carp_address_carries_its_interface_and_state(self):
+        assert _by_address(carp_addresses(OVERVIEW))["192.0.2.1"] == {
+            "mechanism": "carp",
+            "instance": "MGMT",
+            "address": "192.0.2.1",
+            "prefix_length": 24,
+            "interface": "em0",
+            "group_key": "vhid:108",
+            "running": True,
+            "holds": True,
+        }
+
+    def test_an_alias_on_a_vhid_follows_that_vhid(self):
+        alias = _by_address(carp_addresses(OVERVIEW))["198.51.100.140"]
+        assert (alias["group_key"], alias["prefix_length"], alias["holds"]) == (
+            "vhid:140",
+            32,
+            True,
+        )
+
+    def test_a_vhid_not_up_neither_runs_nor_holds(self):
+        entry = _by_address(carp_addresses(OVERVIEW))["203.0.113.1"]
+        assert (entry["running"], entry["holds"]) == (False, False)
+
+    def test_an_interface_without_a_description_is_named_by_its_identifier(self):
+        assert _by_address(carp_addresses(OVERVIEW))["203.0.113.1"]["instance"] == "opt9"
+
+    def test_on_a_backup_node_nothing_is_held(self):
+        backup = json.loads(json.dumps(OVERVIEW).replace('"MASTER"', '"BACKUP"'))
+        entries = carp_addresses(backup)
+        assert entries and all(e["holds"] is False for e in entries)
+
+    def test_the_answer_wrapped_in_an_object_reads_the_same(self):
+        assert carp_addresses({"interfaces": OVERVIEW}) == carp_addresses(OVERVIEW)
+
+    @pytest.mark.parametrize("answer", [[], {}, {"interfaces": None}])
+    def test_a_firewall_without_interfaces_has_none(self, answer):
+        assert carp_addresses(answer) == []
+
+    @pytest.mark.parametrize("answer", ["nonsense", None, 42])
+    def test_an_answer_of_another_shape_is_refused(self, answer):
+        with pytest.raises(ValueError):
+            carp_addresses(answer)
+
+    def test_odd_entries_are_skipped(self):
+        overview = [
+            {"identifier": "opt1", "ipv4": ["192.0.2.9/24", {"vhid": "1"}, None]},
+            "not-an-interface",
+            {"identifier": "opt2", "ipv4": [{"ipaddr": "bogus", "vhid": "2"}]},
+        ]
+        assert carp_addresses(overview) == []
+
+
+class TestGetVirtualIps:
+    def test_a_firewall_reports_carp_and_does_not_look_for_the_linux_mechanisms(self):
+        reading = _driver().get_virtual_ips()
+        assert reading["vip_manager"] is None
+        assert reading["keepalived"] is None
+        assert len(reading["carp"]) == 5
+
+    def test_a_firewall_without_carp_reports_an_empty_list(self):
+        reading = _driver(overview=[OVERVIEW[3], OVERVIEW[4]]).get_virtual_ips()
+        assert reading["carp"] == []
+
+    def test_a_failed_request_is_raised_not_reported_as_none_declared(self):
+        with pytest.raises(RuntimeError):
+            _driver(overview=RuntimeError("API down")).get_virtual_ips()