feat(dhcp): implement DhcpServerMixin against Kea DHCPv4

Fills in the three device-specific methods so the generic diff/apply from
napalm-device-types works against OPNsense: searchReservation for the read,
addReservation/setReservation for the write, service/reconfigure for the
commit.

Until now the driver could only create and delete reservations as a
side-effect of VM provisioning, and never read them back — so there was no
way to see what a firewall already had.

Kea's `subnet` field on a reservation is a model relation that comes back as
the related subnet's CIDR in some versions and as its UUID in others. Both
are accepted and normalised to a CIDR; an unresolvable relation degrades to
an empty string rather than raising, so one orphaned entry cannot make the
whole inventory unreadable.

apply_dhcp_reservation deliberately does not reconfigure: that is the
commit's job, so a batch costs one daemon reload instead of one per entry.

Verified against mocked Kea responses only — no live OPNsense was available
at the time of writing. The CIDR-vs-UUID branch in particular is defensive
rather than empirically confirmed.

napalm_opnsense/opnsense.py

@@ -1440,6 +1440,134 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
 
         return result
 
+    # ------------------------------------------------------------------
+    # DhcpServerMixin implementation (Kea DHCPv4). The diff and the apply
+    # loop are generic and live in napalm_device_types.dhcp; only these
+    # three methods know about Kea's REST shape.
+    # ------------------------------------------------------------------
+
+    def _kea_subnets(self) -> list[dict[str, Any]]:
+        """Return Kea's managed subnets, or raise if the plugin is absent."""
+        try:
+            return self._get("/api/kea/dhcpv4/searchSubnet").get("rows") or []
+        except Exception as exc:
+            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc
+
+    def _kea_subnet_uuid_for(self, subnet: str, ip: str) -> str:
+        """Resolve a reservation's target subnet to Kea's subnet UUID.
+
+        Prefers an exact CIDR match on ``subnet``; falls back to whichever
+        managed subnet contains ``ip`` when the caller left ``subnet`` empty.
+        """
+        rows = self._kea_subnets()
+
+        if subnet:
+            for row in rows:
+                if str(row.get("subnet") or "").strip() == subnet:
+                    return str(row["uuid"])
+            raise ValueError(f"No Kea-managed subnet matches {subnet}")
+
+        ip_obj = ip_address(ip)
+        for row in rows:
+            try:
+                if ip_obj in ip_network(row["subnet"], strict=False):
+                    return str(row["uuid"])
+            except (ValueError, KeyError):
+                continue
+        raise ValueError(f"No Kea-managed subnet contains {ip}")
+
+    def get_dhcp_reservations(self) -> list[dict[str, Any]]:
+        """Return all Kea DHCPv4 static reservations, vendor-neutral.
+
+        This is the *configured* state — ``get_dhcp_leases()`` returns what
+        is actually leased. Kea's ``subnet`` field on a reservation is a
+        model relation: depending on version it comes back as the related
+        subnet's CIDR or as its UUID, so both are accepted and normalised to
+        a CIDR here. An unresolvable relation degrades to an empty string
+        rather than raising — one orphaned reservation must not make the
+        whole inventory unreadable.
+
+        :raises RuntimeError: if the Kea plugin isn't installed/enabled.
+        """
+        try:
+            data = self._post(
+                "/api/kea/dhcpv4/searchReservation",
+                {"current": 1, "rowCount": -1},
+            )
+        except Exception as exc:
+            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc
+
+        rows = data.get("rows") or []
+        if not rows:
+            return []
+
+        cidr_by_uuid = {
+            str(row.get("uuid") or ""): str(row.get("subnet") or "")
+            for row in self._kea_subnets()
+        }
+        known_cidrs = set(cidr_by_uuid.values())
+
+        reservations: list[dict[str, Any]] = []
+        for row in rows:
+            raw_subnet = str(row.get("subnet") or "").strip()
+            if raw_subnet in known_cidrs:
+                subnet = raw_subnet
+            else:
+                subnet = cidr_by_uuid.get(raw_subnet, "")
+
+            reservations.append({
+                "uuid": str(row.get("uuid") or ""),
+                "mac": str(row.get("hw_address") or ""),
+                "ip": str(row.get("ip_address") or ""),
+                "hostname": str(row.get("hostname") or ""),
+                "description": str(row.get("description") or ""),
+                "subnet": subnet,
+            })
+        return reservations
+
+    def apply_dhcp_reservation(
+        self, reservation: dict[str, Any], *, uuid: str | None = None
+    ) -> dict[str, Any]:
+        """Create or update a single Kea DHCPv4 static reservation.
+
+        Does not reconfigure the service — call ``commit_dhcp_reservations()``
+        once after a batch, so a ruleset apply costs one daemon reload rather
+        than one per reservation.
+
+        :raises ValueError: if no Kea-managed subnet matches the reservation.
+        :raises RuntimeError: if Kea rejects the write.
+        """
+        subnet_uuid = self._kea_subnet_uuid_for(
+            str(reservation.get("subnet") or "").strip(),
+            str(reservation.get("ip") or ""),
+        )
+
+        payload = {
+            "reservation": {
+                "subnet": subnet_uuid,
+                "ip_address": reservation.get("ip", ""),
+                "hw_address": reservation.get("mac", ""),
+                "hostname": reservation.get("hostname", ""),
+                "description": reservation.get("description", ""),
+            }
+        }
+        path = (
+            f"/api/kea/dhcpv4/setReservation/{uuid}"
+            if uuid
+            else "/api/kea/dhcpv4/addReservation"
+        )
+        result = self._post(path, payload)
+        if result.get("result") != "saved":
+            raise RuntimeError(
+                f"Kea rejected DHCP reservation for {reservation.get('ip')}: {result}"
+            )
+        return {"success": True}
+
+    def commit_dhcp_reservations(self) -> dict[str, Any]:
+        """Reload Kea so pending reservation writes take effect."""
+        self._post("/api/kea/service/reconfigure")
+        return {"success": True}
+
     def get_services(self) -> list[dict[str, Any]]:
         """Return running services from OPNsense.
 

tests/unit/test_driver.py

@@ -1919,3 +1919,211 @@ class TestCommitFirewallRules:
 
         assert calls == [("/api/firewall/filter/apply", {})]
         assert result == {"status": "ok"}
+
+
+# ---------------------------------------------------------------------------
+# DHCP reservations (Kea) — the vendor-neutral DhcpServerMixin contract
+# ---------------------------------------------------------------------------
+
+KEA_RESV_SUBNETS_RESPONSE = {
+    "rows": [
+        {"uuid": "sub-mgmt", "subnet": "10.10.20.0/24"},
+        {"uuid": "sub-users", "subnet": "10.30.20.0/24"},
+    ]
+}
+
+KEA_RESV_ROWS_RESPONSE = {
+    "rows": [
+        {
+            "uuid": "res-1",
+            "subnet": "10.10.20.0/24",
+            "ip_address": "10.10.20.50",
+            "hw_address": "aa:bb:cc:dd:ee:01",
+            "hostname": "nas",
+            "description": "Home NAS",
+        },
+        {
+            "uuid": "res-2",
+            "subnet": "10.30.20.0/24",
+            "ip_address": "10.30.20.80",
+            "hw_address": "AA:BB:CC:DD:EE:02",
+            "hostname": "",
+            "description": "",
+        },
+    ]
+}
+
+
+class TestGetDhcpReservations:
+    def test_maps_kea_rows_to_vendor_neutral_dicts(self, driver):
+        driver._post = lambda path, data=None: KEA_RESV_ROWS_RESPONSE
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        result = driver.get_dhcp_reservations()
+
+        assert result[0] == {
+            "uuid": "res-1",
+            "mac": "aa:bb:cc:dd:ee:01",
+            "ip": "10.10.20.50",
+            "hostname": "nas",
+            "description": "Home NAS",
+            "subnet": "10.10.20.0/24",
+        }
+        assert result[1]["hostname"] == ""
+        assert result[1]["description"] == ""
+
+    def test_requests_all_rows_not_just_first_page(self, driver):
+        calls = []
+        driver._post = lambda path, data=None: calls.append((path, data)) or KEA_RESV_ROWS_RESPONSE
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        driver.get_dhcp_reservations()
+
+        assert calls[0][0] == "/api/kea/dhcpv4/searchReservation"
+        assert calls[0][1]["rowCount"] == -1
+
+    def test_resolves_subnet_uuid_to_cidr(self, driver):
+        # Depending on version, searchReservation returns the relation's UUID
+        # rather than its display value — the caller must still get a CIDR.
+        driver._post = lambda path, data=None: {
+            "rows": [
+                {
+                    "uuid": "res-1",
+                    "subnet": "sub-users",
+                    "ip_address": "10.30.20.50",
+                    "hw_address": "aa:bb:cc:dd:ee:03",
+                    "hostname": "printer",
+                    "description": "",
+                }
+            ]
+        }
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        result = driver.get_dhcp_reservations()
+
+        assert result[0]["subnet"] == "10.30.20.0/24"
+
+    def test_unresolvable_subnet_degrades_to_empty_string(self, driver):
+        driver._post = lambda path, data=None: {
+            "rows": [
+                {
+                    "uuid": "res-1",
+                    "subnet": "sub-gone",
+                    "ip_address": "10.99.0.5",
+                    "hw_address": "aa:bb:cc:dd:ee:04",
+                    "hostname": "",
+                    "description": "",
+                }
+            ]
+        }
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        assert driver.get_dhcp_reservations()[0]["subnet"] == ""
+
+    def test_empty_device_returns_empty_list(self, driver):
+        driver._post = lambda path, data=None: {"rows": []}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        assert driver.get_dhcp_reservations() == []
+
+    def test_kea_plugin_unavailable_raises_runtime_error(self, driver):
+        def _boom(path, data=None):
+            raise RuntimeError("404 Not Found")
+
+        driver._post = _boom
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
+            driver.get_dhcp_reservations()
+
+
+class TestApplyDhcpReservation:
+    def _reservation(self, **overrides):
+        base = {
+            "uuid": "",
+            "mac": "aa:bb:cc:dd:ee:01",
+            "ip": "10.10.20.50",
+            "hostname": "nas",
+            "description": "Home NAS",
+            "subnet": "10.10.20.0/24",
+        }
+        base.update(overrides)
+        return base
+
+    def test_add_posts_to_add_reservation(self, driver):
+        calls = []
+        driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        result = driver.apply_dhcp_reservation(self._reservation())
+
+        assert result == {"success": True}
+        assert calls[0][0] == "/api/kea/dhcpv4/addReservation"
+        assert calls[0][1] == {
+            "reservation": {
+                "subnet": "sub-mgmt",
+                "ip_address": "10.10.20.50",
+                "hw_address": "aa:bb:cc:dd:ee:01",
+                "hostname": "nas",
+                "description": "Home NAS",
+            }
+        }
+
+    def test_update_posts_to_set_reservation_with_uuid(self, driver):
+        calls = []
+        driver._post = lambda path, data=None: calls.append(path) or {"result": "saved"}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        driver.apply_dhcp_reservation(self._reservation(), uuid="res-1")
+
+        assert calls == ["/api/kea/dhcpv4/setReservation/res-1"]
+
+    def test_does_not_reconfigure_on_every_write(self, driver):
+        # Committing is a separate step — one reconfigure per ruleset, not per
+        # reservation.
+        calls = []
+        driver._post = lambda path, data=None: calls.append(path) or {"result": "saved"}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        driver.apply_dhcp_reservation(self._reservation())
+
+        assert not any("reconfigure" in path for path in calls)
+
+    def test_unknown_subnet_raises_value_error(self, driver):
+        driver._post = lambda path, data=None: {"result": "saved"}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        with pytest.raises(ValueError, match="No Kea-managed subnet"):
+            driver.apply_dhcp_reservation(self._reservation(subnet="192.168.99.0/24"))
+
+    def test_falls_back_to_containing_subnet_when_cidr_absent(self, driver):
+        # A caller that only knows the IP (no subnet field) should still work:
+        # Kea's own subnet list decides which subnet the address belongs to.
+        calls = []
+        driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        driver.apply_dhcp_reservation(self._reservation(subnet=""))
+
+        assert calls[0][1]["reservation"]["subnet"] == "sub-mgmt"
+
+    def test_rejected_write_raises_runtime_error(self, driver):
+        driver._post = lambda path, data=None: {
+            "result": "failed",
+            "validations": {"reservation.ip_address": "Duplicate entry exists."},
+        }
+        driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
+
+        with pytest.raises(RuntimeError, match="Kea rejected"):
+            driver.apply_dhcp_reservation(self._reservation())
+
+
+class TestCommitDhcpReservations:
+    def test_posts_reconfigure(self, driver):
+        calls = []
+        driver._post = lambda path, data=None: calls.append(path) or {"status": "ok"}
+
+        result = driver.commit_dhcp_reservations()
+
+        assert calls == ["/api/kea/service/reconfigure"]
+        assert result == {"success": True}