From 806a23018d1ebdd40c893b244ccdf6e82b4d3a3e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 8 Jul 2026 09:09:41 +0200 Subject: [PATCH] feat(opnsense): add create_dhcp_reservation() for Kea DHCPv4 static mappings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Only Kea (os-kea plugin) is supported — no active OPNsense environment with legacy ISC DHCP was available to verify a second code path against. Payload/response shapes (searchSubnet, searchReservation, addReservation, setReservation, delReservation, service/reconfigure) were confirmed against a real OPNsense box via a live add + verify + delete cycle before writing this method and its tests. --- napalm_opnsense/opnsense.py | 73 ++++++++++++++++++++++++++++ tests/unit/test_driver.py | 97 +++++++++++++++++++++++++++++++++++++ 2 files changed, 170 insertions(+) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 0d8e941..df70ba7 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -39,6 +39,7 @@ import difflib import json import logging import socket +from ipaddress import ip_address, ip_network from typing import Any logger = logging.getLogger(__name__) @@ -1135,6 +1136,78 @@ class OPNsenseDriver(FirewallDriver): logger.warning("ARP table fallback failed: %s", exc) return [] + def create_dhcp_reservation(self, mac: str, ip: str, hostname: str = "") -> None: + """Create (or update) a Kea DHCPv4 static reservation (MAC → IP). + + Only the Kea backend (os-kea plugin) is supported — this driver has + no active OPNsense environment with legacy ISC DHCP to verify a + second code path against, unlike ``get_dhcp_leases()``'s read-only + try-then-fallback. Callers should treat a missing/disabled Kea + plugin as "reservations unsupported" (``RuntimeError``), not fall + back to plain DHCP silently. + + :param mac: NIC MAC address (any common formatting; sent as-is to + Kea's ``hw_address`` field). + :param ip: IP address to reserve; must fall inside a subnet Kea + already manages (``searchSubnet``), or this raises ValueError. + :param hostname: optional hostname to record on the reservation. + :raises ValueError: if no Kea subnet contains ``ip``. + :raises RuntimeError: if Kea rejects the reservation (validation + errors) or the Kea plugin isn't installed/enabled. + """ + try: + subnets = self._get("/api/kea/dhcpv4/searchSubnet").get("rows") or [] + except Exception as exc: + raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc + + ip_obj = ip_address(ip) + subnet_uuid = None + for row in subnets: + try: + if ip_obj in ip_network(row["subnet"], strict=False): + subnet_uuid = row["uuid"] + break + except ValueError: + continue + if subnet_uuid is None: + raise ValueError(f"No Kea-managed subnet contains {ip}") + + # Idempotency: reuse an existing reservation for this IP if one + # already exists (e.g. a retried provisioning job), rather than + # creating a duplicate Kea rejects anyway. + existing_uuid = None + try: + existing = self._post( + "/api/kea/dhcpv4/searchReservation", + {"current": 1, "rowCount": -1, "searchPhrase": ip}, + ) + for row in existing.get("rows") or []: + if row.get("ip_address") == ip: + existing_uuid = row.get("uuid") + break + except Exception as exc: + logger.debug("Kea reservation search failed, proceeding to add: %s", exc) + + payload = { + "reservation": { + "subnet": subnet_uuid, + "ip_address": ip, + "hw_address": mac, + "hostname": hostname, + "description": self._NETORK_TAG, + } + } + path = ( + f"/api/kea/dhcpv4/setReservation/{existing_uuid}" + if existing_uuid + else "/api/kea/dhcpv4/addReservation" + ) + result = self._post(path, payload) + if result.get("result") != "saved": + raise RuntimeError(f"Kea rejected DHCP reservation for {ip}: {result}") + + self._post("/api/kea/service/reconfigure") + 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 c928af1..044b86a 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1243,3 +1243,100 @@ class TestGetConfigCandidate: result = driver.get_config() parsed = json.loads(result["candidate"]) assert isinstance(parsed, list) + + +# --------------------------------------------------------------------------- +# create_dhcp_reservation() +# +# Payload/response shapes below are taken verbatim from a live probe against +# a real OPNsense box running the Kea DHCPv4 (os-kea) plugin — searchSubnet, +# searchReservation, addReservation, delReservation, and service/reconfigure +# were all exercised live (including a real add + verify + delete cycle) to +# confirm the exact request/response schema before writing this driver +# method and these tests against it. +# --------------------------------------------------------------------------- + +KEA_SUBNETS_RESPONSE = { + "rows": [ + {"uuid": "82766878-c5ac-41f3-b7b0-e24d2419beb3", "subnet": "172.22.0.0/24"}, + {"uuid": "6854cab3-ebb5-4031-b987-0edcc6723546", "subnet": "172.22.8.0/24"}, + ] +} + + +class TestCreateDhcpReservation: + def test_adds_new_reservation_when_none_exists(self, driver): + driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE) + driver.session.post.side_effect = [ + _make_json_response({"rows": []}), # searchReservation — no match + _make_json_response({"result": "saved", "uuid": "new-uuid-123"}), # addReservation + _make_json_response({"status": "ok"}), # service/reconfigure + ] + + driver.create_dhcp_reservation( + mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253", hostname="new-vm" + ) + + add_call = driver.session.post.call_args_list[1] + assert add_call.args[0] == "https://opnsense.example.com/api/kea/dhcpv4/addReservation" + payload = add_call.kwargs["json"]["reservation"] + assert payload["subnet"] == "6854cab3-ebb5-4031-b987-0edcc6723546" + assert payload["ip_address"] == "172.22.8.253" + assert payload["hw_address"] == "02:aa:bb:cc:dd:ee" + assert payload["hostname"] == "new-vm" + assert payload["description"] == "[netork]" + + reconfigure_call = driver.session.post.call_args_list[2] + assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + + def test_updates_existing_reservation_for_same_ip(self, driver): + driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE) + driver.session.post.side_effect = [ + _make_json_response( + {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]} + ), + _make_json_response({"result": "saved", "uuid": "existing-uuid-456"}), + _make_json_response({"status": "ok"}), + ] + + driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253") + + set_call = driver.session.post.call_args_list[1] + assert ( + set_call.args[0] + == "https://opnsense.example.com/api/kea/dhcpv4/setReservation/existing-uuid-456" + ) + + def test_raises_when_ip_not_in_any_kea_subnet(self, driver): + driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE) + + with pytest.raises(ValueError, match="No Kea-managed subnet"): + driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="10.99.99.99") + + driver.session.post.assert_not_called() + + def test_raises_when_kea_rejects_reservation(self, driver): + driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE) + driver.session.post.side_effect = [ + _make_json_response({"rows": []}), + _make_json_response( + { + "result": "failed", + "validations": {"reservation.ip_address": "Address not in specified subnet"}, + } + ), + ] + + with pytest.raises(RuntimeError, match="Kea rejected"): + driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253") + + # reconfigure must NOT be called after a rejected reservation + assert driver.session.post.call_count == 2 + + def test_raises_when_kea_plugin_unavailable(self, driver): + driver.session.get.side_effect = Exception("404 Not Found") + + with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"): + driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253") + + driver.session.post.assert_not_called()