fix: name and version from system_information; a CARP address on its MASTER's interface

get_facts read hostname and version from /api/core/system/status, which on
OPNsense 26.7 carries neither: every firewall read as nameless and "unknown",
and netOrk kept naming one by its address. /api/diagnostics/system/system_information
has both (name is the FQDN, versions[0] "OPNsense 26.7.7-amd64"). The status
answer still counts where the new one gives nothing.

get_interfaces_ip listed each interface's primary address only, so a CARP address
never showed as the holder's. It now lists an address bound to a VHID on the node
that is MASTER for it, and leaves it off a BACKUP -- as keepalived puts the address
on the master alone. netOrk's VIP badge sits on that interface (netOrk #1121).

README.md

@@ -71,9 +71,9 @@ arguments.
 
 | Getter | Status | OPNsense API Endpoint |
 |---|---|---|
-| `get_facts` | ✅ | `GET /api/core/system/status` |
+| `get_facts` | ✅ | `GET /api/diagnostics/system/system_information` (name, version) + `GET /api/core/system/status` (fallback on older versions) |
 | `get_interfaces` | ✅ | `GET /api/interfaces/overview/export` |
-| `get_interfaces_ip` | ✅ | `GET /api/interfaces/addresses/export` |
+| `get_interfaces_ip` | ✅ | `GET /api/interfaces/overview/export`; a CARP address only on the node that is MASTER for it |
 | `get_interfaces_counters` | ✅ | `GET /api/diagnostics/interface/get_interface_statistics` |
 | `get_arp_table` | ✅ | `GET /api/diagnostics/interface/get_arp` |
 | `get_ipv6_neighbors_table` | ✅ | `GET /api/diagnostics/interface/get_ndp` |

napalm_opnsense/opnsense.py

@@ -38,6 +38,7 @@ from __future__ import annotations
 import difflib
 import json
 import logging
+import re
 import socket
 import time
 from ipaddress import ip_address, ip_network
@@ -68,6 +69,23 @@ from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interf
 from napalm_opnsense.virtual_ips import carp_addresses
 
 
+_ARCHITECTURE_SUFFIX = re.compile(r"-(amd64|aarch64|arm64|i386|armv7)$")
+
+
+def _opnsense_version(versions: Any) -> str:
+    """``["OPNsense 26.7.7-amd64", "FreeBSD ..."]`` -> ``"26.7.7"``; "" when absent."""
+    for line in versions if isinstance(versions, list) else []:
+        product, _, version = str(line).partition(" ")
+        if product == "OPNsense" and version:
+            return _ARCHITECTURE_SUFFIX.sub("", version.strip())
+    return ""
+
+
+def _carp_master(entry: dict[str, Any]) -> bool:
+    """Whether this node is MASTER for an interface address bound to a CARP VHID."""
+    return str(entry.get("status") or "").strip().upper() == "MASTER"
+
+
 class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
     """NAPALM driver for OPNsense (read-only, REST API)."""
 
@@ -217,16 +235,33 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
     def get_facts(self) -> dict[str, Any]:
         """Return a dictionary of general device facts.
 
-        Calls ``GET /api/core/system/status``.
+        Name and version come from ``GET /api/diagnostics/system/system_information``
+        (``name`` is the FQDN, ``versions[0]`` reads ``"OPNsense 26.7.7-amd64"``):
+        ``/api/core/system/status`` carries neither on OPNsense 26.7. The status
+        answer is still read, and still counts where the other one gives nothing --
+        an older OPNsense, or an API key without the diagnostics privilege.
 
         Returned keys (NAPALM standard):
         ``vendor``, ``model``, ``hostname``, ``fqdn``, ``os_version``,
         ``serial_number``, ``uptime``, ``interface_list``.
         """
         status = self._get("/api/core/system/status")
+        try:
+            information = self._get("/api/diagnostics/system/system_information")
+        except Exception as exc:
+            logger.debug("system_information not read: %s", exc)
+            information = {}
+        if not isinstance(information, dict):
+            information = {}
 
-        hostname = status.get("hostname") or status.get("name") or ""
+        fqdn = str(information.get("name") or status.get("hostname") or status.get("name") or "")
-        version = status.get("version") or status.get("product_version") or "unknown"
+        hostname = fqdn.split(".", 1)[0]
+        version = (
+            _opnsense_version(information.get("versions"))
+            or status.get("version")
+            or status.get("product_version")
+            or "unknown"
+        )
 
         try:
             interface_list = list(self.get_interfaces().keys())
@@ -238,7 +273,7 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
             "vendor": self.VENDOR,
             "model": status.get("model") or self.VENDOR,
             "hostname": hostname,
-            "fqdn": hostname,
+            "fqdn": fqdn,
             "os_version": version,
             "serial_number": status.get("serial") or "",
             "uptime": status.get("uptime", -1),
@@ -288,6 +323,10 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
 
         Uses ``GET /api/interfaces/overview/export`` (same source as get_interfaces).
 
+        An address bound to a CARP VHID is listed on every node; it counts as this
+        node's only while the node is its MASTER, as keepalived puts the address on
+        the master alone (netOrk #1121).
+
         Structure::
 
             {
@@ -323,8 +362,14 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
                 _add(name, addr4, "ipv4")
             if addr6:
                 _add(name, addr6, "ipv6")
+            for family in ("ipv4", "ipv6"):
+                for entry in iface.get(family) or []:
+                    if isinstance(entry, dict) and entry.get("vhid") and _carp_master(entry):
+                        _add(name, entry.get("ipaddr") or "", family)
             if not addr4:
                 for entry in iface.get("ipv4") or []:
+                    if entry.get("vhid") and not _carp_master(entry):
+                        continue
                     ip_field = entry.get("ipaddr") or entry.get("ip", "")
                     subnet = entry.get("subnetbits") or entry.get("prefix_length")
                     cidr = f"{ip_field}/{subnet}" if subnet and "/" not in ip_field else ip_field
@@ -332,6 +377,8 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver):
                         _add(name, cidr, "ipv4")
             if not addr6:
                 for entry in iface.get("ipv6") or []:
+                    if entry.get("vhid") and not _carp_master(entry):
+                        continue
                     ip_field = entry.get("ipaddr") or entry.get("ip", "")
                     prefix = entry.get("prefixlen") or entry.get("prefix_length")
                     cidr = f"{ip_field}/{prefix}" if prefix and "/" not in ip_field else ip_field

tests/unit/test_identity_and_held_addresses.py

@@ -0,0 +1,132 @@
+"""What the firewall is called, which version it runs, and which addresses it serves.
+
+``/api/core/system/status`` on OPNsense 26.7 carries neither name nor version
+(``metadata`` and ``subsystems`` only), so every firewall read as nameless and
+"unknown". ``/api/diagnostics/system/system_information`` has both: ``name`` is
+the FQDN, ``versions[0]`` reads ``"OPNsense 26.7.7-amd64"``.
+
+An address bound to a CARP VHID sits on the interface of every node, in state
+MASTER or BACKUP. Only the MASTER answers for it, so only there is it one of the
+node's addresses -- the same as keepalived, where only the master has the address
+at all (netOrk #1121: a virtual IP's badge sits on its holder's interface).
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+OVERVIEW = json.loads(
+    (Path(__file__).parent / "fixtures" / "interfaces_overview_carp.json").read_text()
+)
+SYSTEM_INFORMATION = {
+    "name": "gw-02.home.manivong.de",
+    "versions": ["OPNsense 26.7.7-amd64", "FreeBSD 15.1-RELEASE-p4", "OpenSSL 3.5.9"],
+}
+STATUS_26_7 = {"metadata": {"translations": {}}, "subsystems": {}}
+
+
+@pytest.fixture
+def driver() -> OPNsenseDriver:
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        return OPNsenseDriver(hostname="fw", username="k", password="s")
+
+
+def _answers(driver, system_information=SYSTEM_INFORMATION, status=STATUS_26_7, overview=OVERVIEW):
+    def get(path: str):
+        if path == "/api/diagnostics/system/system_information":
+            if isinstance(system_information, Exception):
+                raise system_information
+            return system_information
+        if path == "/api/core/system/status":
+            return status
+        if path == "/api/interfaces/overview/export":
+            return overview
+        raise AssertionError(f"unexpected GET {path}")
+
+    driver._get = get
+    return driver
+
+
+class TestIdentity:
+    def test_the_name_is_the_short_host_and_the_fqdn_the_whole(self, driver):
+        facts = _answers(driver).get_facts()
+        assert (facts["hostname"], facts["fqdn"]) == ("gw-02", "gw-02.home.manivong.de")
+
+    def test_the_version_comes_without_the_architecture(self, driver):
+        assert _answers(driver).get_facts()["os_version"] == "26.7.7"
+
+    @pytest.mark.parametrize(
+        "line, version",
+        [
+            ("OPNsense 26.7.7_1-aarch64", "26.7.7_1"),
+            ("OPNsense 25.1", "25.1"),
+            ("OPNsense 24.7.12-i386", "24.7.12"),
+        ],
+    )
+    def test_version_spellings(self, driver, line, version):
+        info = {"name": "fw", "versions": [line]}
+        assert _answers(driver, system_information=info).get_facts()["os_version"] == version
+
+    def test_a_name_without_a_domain_is_both(self, driver):
+        facts = _answers(driver, system_information={"name": "fw", "versions": []}).get_facts()
+        assert (facts["hostname"], facts["fqdn"]) == ("fw", "fw")
+
+    def test_without_system_information_the_status_answer_still_counts(self, driver):
+        """An older OPNsense, or an API key without the diagnostics privilege."""
+        status = {"hostname": "opnsense01", "version": "24.7"}
+        facts = _answers(driver, system_information=RuntimeError("403"), status=status).get_facts()
+        assert (facts["hostname"], facts["os_version"]) == ("opnsense01", "24.7")
+
+    def test_nothing_known_is_empty_and_unknown(self, driver):
+        facts = _answers(driver, system_information={}).get_facts()
+        assert (facts["hostname"], facts["fqdn"], facts["os_version"]) == ("", "", "unknown")
+
+
+class TestHeldAddresses:
+    def test_the_master_serves_its_carp_addresses(self, driver):
+        result = _answers(driver).get_interfaces_ip()
+        assert set(result["em0"]["ipv4"]) == {"192.0.2.2", "192.0.2.1"}
+        assert result["em0"]["ipv4"]["192.0.2.1"] == {"prefix_length": 24}
+        assert set(result["vlan0040"]["ipv4"]) == {
+            "198.51.100.2",
+            "198.51.100.1",
+            "198.51.100.140",
+        }
+
+    def test_a_backup_serves_its_own_address_only(self, driver):
+        backup = json.loads(json.dumps(OVERVIEW).replace('"MASTER"', '"BACKUP"'))
+        result = _answers(driver, overview=backup).get_interfaces_ip()
+        assert set(result["em0"]["ipv4"]) == {"192.0.2.2"}
+        assert set(result["vlan0040"]["ipv4"]) == {"198.51.100.2"}
+
+    def test_a_vhid_that_is_not_up_is_not_served(self, driver):
+        result = _answers(driver).get_interfaces_ip()
+        assert set(result["vlan0011"]["ipv4"]) == {"203.0.113.2"}
+
+    def test_a_held_ipv6_carp_address_is_served(self, driver):
+        result = _answers(driver).get_interfaces_ip()
+        assert "2001:db8:8::1" in result["em0"]["ipv6"]
+
+    def test_an_address_without_a_vhid_is_left_as_before(self, driver):
+        """Only CARP addresses are added; a plain alias is a separate question."""
+        result = _answers(driver).get_interfaces_ip()
+        assert set(result["vlan0020"]["ipv4"]) == {"192.0.2.130"}
+
+    def test_without_a_primary_address_the_list_still_leaves_out_a_backup(self, driver):
+        overview = [
+            {
+                "device": "vlan0099",
+                "ipv4": [
+                    {"ipaddr": "192.0.2.66/26"},
+                    {"ipaddr": "192.0.2.65/26", "vhid": "199", "status": "BACKUP"},
+                ],
+            }
+        ]
+        result = _answers(driver, overview=overview).get_interfaces_ip()
+        assert set(result["vlan0099"]["ipv4"]) == {"192.0.2.66"}