fix(dhcp): read the Kea subnet record from subnet4, not subnet
CI / test (3.10) (push) Failing after 8s
CI / test (3.11) (push) Failing after 8s
CI / test (3.12) (push) Failing after 13s
CI / test (3.9) (push) Failing after 8s

getSubnet wraps its record under `subnet4`. The driver read `subnet`, got
nothing, and carried on:

  - get_dhcp_subnets() reported every subnet with no pools, no options and no
    description. Only the CIDR survived, and only because it falls back to the
    searchSubnet row. Confirmed against a live OPNsense serving six subnets:
    all six came back with empty pools while the device had
    "10.10.0.100-10.10.0.250" and routers/DNS/NTP set on each.

  - apply_dhcp_subnet() read the same key to merge the options it was not
    asked to change. An empty record means nothing to preserve, so updating a
    subnet with only domain_search set would have written back only that one
    option and blanked the routers Kea autocollected — stranding every client
    on that VLAN without a gateway. That is precisely the failure the merge
    exists to prevent.

The unit fixtures encoded the wrong shape, which is why the safety test
test_unnamed_options_are_preserved_on_update passed while the real thing was
broken. They now carry the response captured from OPNsense 25.x, and correcting
them turns that test red against the old parse.

Both call sites go through _kea_subnet_record(), which prefers `subnet4` and
falls back to `subnet` for older builds.
This commit is contained in:
Christian Manivong
2026-08-20 11:49:16 +07:00
parent 0c5670981d
commit 960aaefa13
2 changed files with 40 additions and 7 deletions
+16 -2
View File
@@ -1618,6 +1618,20 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
return selected[0] if selected else "" return selected[0] if selected else ""
return str(raw or "").strip() return str(raw or "").strip()
@staticmethod
def _kea_subnet_record(detail: dict[str, Any]) -> dict[str, Any]:
"""The subnet record out of a ``getSubnet`` response.
OPNsense wraps it under ``subnet4``; older builds used ``subnet``, and
keeping the fallback costs nothing. Reading the wrong key costs a
great deal: the record comes back empty, so a subnet reports no pools
and no options at all, and on update the partial-option merge has
nothing to preserve and blanks every option Kea autocollected --
stranding a whole VLAN without a gateway, which is the exact failure
that merge exists to prevent.
"""
return detail.get("subnet4") or detail.get("subnet") or {}
def get_dhcp_subnets(self) -> list[dict[str, Any]]: def get_dhcp_subnets(self) -> list[dict[str, Any]]:
"""Return every Kea DHCPv4 subnet with its pools and options. """Return every Kea DHCPv4 subnet with its pools and options.
@@ -1650,7 +1664,7 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
logger.warning("Kea subnet %s detail fetch failed, skipping: %s", uuid, exc) logger.warning("Kea subnet %s detail fetch failed, skipping: %s", uuid, exc)
continue continue
record = detail.get("subnet") or {} record = self._kea_subnet_record(detail)
raw_options = record.get("option_data") or {} raw_options = record.get("option_data") or {}
option_data: dict[str, Any] = {} option_data: dict[str, Any] = {}
@@ -1706,7 +1720,7 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
if uuid and desired_options: if uuid and desired_options:
try: try:
current = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}") current = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}")
raw = (current.get("subnet") or {}).get("option_data") or {} raw = self._kea_subnet_record(current).get("option_data") or {}
except Exception as exc: except Exception as exc:
raise RuntimeError( raise RuntimeError(
f"Kea subnet {uuid} could not be read before update: {exc}" f"Kea subnet {uuid} could not be read before update: {exc}"
+24 -5
View File
@@ -2177,10 +2177,11 @@ KEA_SUBNET_SEARCH_RESPONSE = {
] ]
} }
# getSubnet wraps the record and, unlike searchSubnet, carries the options. # getSubnet wraps the record under `subnet4` -- captured from OPNsense 25.x,
# OPNsense renders AsList fields as comma-separated strings. # not guessed. Unlike searchSubnet it carries the options, and OPNsense
# renders AsList fields as comma-separated strings.
KEA_SUBNET_DETAIL_RESPONSE = { KEA_SUBNET_DETAIL_RESPONSE = {
"subnet": { "subnet4": {
"subnet": "10.10.20.0/24", "subnet": "10.10.20.0/24",
"description": "Home", "description": "Home",
"pools": "10.10.20.100-10.10.20.200", "pools": "10.10.20.100-10.10.20.200",
@@ -2243,7 +2244,7 @@ class TestGetDhcpSubnets:
def test_multiline_pools_become_a_list(self, driver): def test_multiline_pools_become_a_list(self, driver):
detail = { detail = {
"subnet": { "subnet4": {
"subnet": "10.10.20.0/24", "subnet": "10.10.20.0/24",
"description": "", "description": "",
"pools": "10.10.20.100-10.10.20.150\n10.10.20.180-10.10.20.200", "pools": "10.10.20.100-10.10.20.150\n10.10.20.180-10.10.20.200",
@@ -2259,7 +2260,7 @@ class TestGetDhcpSubnets:
def test_selection_map_shape_is_accepted(self, driver): def test_selection_map_shape_is_accepted(self, driver):
"""Some OPNsense versions render list fields as a selection map.""" """Some OPNsense versions render list fields as a selection map."""
detail = { detail = {
"subnet": { "subnet4": {
"subnet": "10.10.20.0/24", "subnet": "10.10.20.0/24",
"description": "", "description": "",
"pools": "", "pools": "",
@@ -2400,3 +2401,21 @@ class TestCommitDhcpSubnets:
assert driver.commit_dhcp_subnets() == {"success": True} assert driver.commit_dhcp_subnets() == {"success": True}
assert calls == ["/api/kea/service/reconfigure"] assert calls == ["/api/kea/service/reconfigure"]
class TestKeaSubnetRecordKey:
"""`getSubnet` wraps its record under `subnet4`. Reading the wrong key is
not a cosmetic miss: the record comes back empty, so every subnet reports
no pools and no options, and the partial-option merge on update has
nothing to preserve."""
def test_reads_the_subnet4_wrapper(self, driver):
record = driver._kea_subnet_record({"subnet4": {"subnet": "10.0.0.0/24"}})
assert record == {"subnet": "10.0.0.0/24"}
def test_falls_back_to_the_legacy_subnet_wrapper(self, driver):
record = driver._kea_subnet_record({"subnet": {"subnet": "10.0.0.0/24"}})
assert record == {"subnet": "10.0.0.0/24"}
def test_unknown_shape_yields_an_empty_record(self, driver):
assert driver._kea_subnet_record({"something_else": {}}) == {}