feat(opnsense): add create_dhcp_reservation() for Kea DHCPv4 static mappings

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.
This commit is contained in:
Christian Manivong
2026-07-08 09:09:41 +02:00
parent e3c36a1d34
commit 806a23018d
2 changed files with 170 additions and 0 deletions
+73
View File
@@ -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.