From 1eb378c04c4ad1b4087a46a60d38a10284c4b612 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 19 Aug 2026 07:24:48 +0700 Subject: [PATCH] feat(dhcp): implement DhcpServerMixin against Kea DHCPv4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 128 ++++++++++++++++++++++ tests/unit/test_driver.py | 208 ++++++++++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index d24077f..a47b4a7 100644 --- a/napalm_opnsense/opnsense.py +++ b/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. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 0a95a56..0932749 100644 --- a/tests/unit/test_driver.py +++ b/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}