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,124 @@
+"""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 CARP protocol
+knowledge, so it lives here with the driver that speaks CARP
+(:func:`carp_virtual_ip`), not in napalm-device-types, which carries only the
+shape of the reading.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+from typing import Any, Optional
+
+from napalm_device_types.models import VirtualIpDict
+
+#: CARP's state of a VHID on one member -> (running, holds).
+_CARP_STATES: dict[str, tuple[bool, bool]] = {
+    "MASTER": (True, True),
+    "BACKUP": (True, False),
+    "INIT": (False, False),
+    "DISABLED": (False, False),
+}
+
+
+def _address(text: str) -> Optional[tuple[str, Optional[int]]]:
+    """``"10.0.0.1/24"`` -> ``("10.0.0.1", 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 carp_virtual_ip(
+    *,
+    address: str,
+    prefix_length: Optional[int],
+    vhid: object,
+    state: Optional[str],
+    instance: str,
+    interface: Optional[str],
+) -> Optional[VirtualIpDict]:
+    """One address bound to a CARP VHID, as a virtual IP; None for what is no address.
+
+    Who holds it comes from CARP's *state* on this member, not from its
+    interfaces: MASTER holds it, BACKUP stands by, INIT and DISABLED neither run
+    nor hold. A state this does not know says nothing (``running`` and ``holds``
+    None), and the caller falls back on the interfaces. *address* may carry its
+    prefix (``"10.0.0.1/24"``), which then wins over *prefix_length*.
+    """
+    parsed = _address(address.strip()) if address else None
+    if parsed is None:
+        return None
+    normalised, written = parsed
+    prefix = written if written is not None else prefix_length
+    if prefix is not None and not 0 <= prefix <= ipaddress.ip_address(normalised).max_prefixlen:
+        prefix = None
+    vhid_text = str(vhid).strip() if vhid is not None else ""
+    running, holds = _CARP_STATES.get((state or "").strip().upper(), (None, None))
+    return {
+        "mechanism": "carp",
+        "instance": instance or f"vhid {vhid_text}",
+        "address": normalised,
+        "prefix_length": prefix,
+        "interface": interface,
+        "group_key": f"vhid:{vhid_text}" if vhid_text.isdigit() else None,
+        "running": running,
+        "holds": holds,
+    }
+
+
+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.1",
     "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_carp_states.py

@@ -0,0 +1,111 @@
+"""carp_virtual_ip: one CARP address as a virtual IP, with its holder in the reading.
+
+CARP differs from vip-manager and keepalived in what the caller can see. The
+address is configured on every member, and the interface list of a firewall need
+not show it at all -- so the interfaces cannot say who holds it. CARP's own state
+can: the member in MASTER holds it, a BACKUP stands by. That is CARP protocol
+knowledge, so it lives with the driver that speaks CARP (netOrk #1121), not in
+napalm-device-types.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_opnsense.virtual_ips import carp_virtual_ip
+
+
+def _entry(**overrides):
+    fields = {
+        "address": "172.22.8.1",
+        "prefix_length": 24,
+        "vhid": "108",
+        "state": "MASTER",
+        "instance": "MGMT",
+        "interface": "vtnet0",
+    }
+    fields.update(overrides)
+    return carp_virtual_ip(**fields)
+
+
+def test_a_master_holds_the_address():
+    assert _entry() == {
+        "mechanism": "carp",
+        "instance": "MGMT",
+        "address": "172.22.8.1",
+        "prefix_length": 24,
+        "interface": "vtnet0",
+        "group_key": "vhid:108",
+        "running": True,
+        "holds": True,
+    }
+
+
+def test_a_backup_stands_by():
+    entry = _entry(state="BACKUP")
+    assert entry is not None
+    assert (entry["running"], entry["holds"]) == (True, False)
+
+
+@pytest.mark.parametrize("state", ["INIT", "DISABLED"])
+def test_a_member_whose_carp_is_not_up_neither_runs_nor_holds(state):
+    entry = _entry(state=state)
+    assert entry is not None
+    assert (entry["running"], entry["holds"]) == (False, False)
+
+
+@pytest.mark.parametrize("state", [None, "", "SOMETHING-NEW"])
+def test_an_unknown_state_says_nothing_about_the_holder(state):
+    """None leaves the holder to the interfaces, as for the other mechanisms."""
+    entry = _entry(state=state)
+    assert entry is not None
+    assert (entry["running"], entry["holds"]) == (None, None)
+
+
+def test_the_state_is_read_case_blind():
+    entry = _entry(state=" master ")
+    assert entry is not None
+    assert entry["holds"] is True
+
+
+def test_the_address_is_normalised():
+    entry = _entry(address="FD00::0001", prefix_length=64)
+    assert entry is not None
+    assert entry["address"] == "fd00::1"
+
+
+def test_an_address_written_with_its_prefix_takes_that_prefix():
+    entry = _entry(address="10.42.100.1/24", prefix_length=None)
+    assert entry is not None
+    assert (entry["address"], entry["prefix_length"]) == ("10.42.100.1", 24)
+
+
+@pytest.mark.parametrize("address", ["", "not-an-address", "10.0.0.300"])
+def test_what_is_no_address_is_no_virtual_ip(address):
+    assert _entry(address=address) is None
+
+
+@pytest.mark.parametrize("prefix", [33, -1])
+def test_an_impossible_prefix_is_dropped(prefix):
+    entry = _entry(prefix_length=prefix)
+    assert entry is not None
+    assert entry["prefix_length"] is None
+
+
+@pytest.mark.parametrize("vhid", [None, "", "abc"])
+def test_without_a_numeric_vhid_there_is_no_group(vhid):
+    entry = _entry(vhid=vhid)
+    assert entry is not None
+    assert entry["group_key"] is None
+
+
+def test_a_vhid_given_as_a_number_groups_the_same_way():
+    entry = _entry(vhid=108)
+    assert entry is not None
+    assert entry["group_key"] == "vhid:108"
+
+
+def test_an_empty_instance_falls_back_to_the_vhid():
+    entry = _entry(instance="")
+    assert entry is not None
+    assert entry["instance"] == "vhid 108"

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()