From b8dac1db63f08d95de77fa00b594286635d14055 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 8 Jul 2026 19:24:04 +0200 Subject: [PATCH] feat(opnsense): add delete_dhcp_reservation_and_lease() for Kea DHCPv4 Combined removal of a static reservation and its active lease, needed by NetOrk's VM-deletion cleanup flow. Reservation deletion follows the same search-then-del/{uuid} + reconfigure pattern as create_dhcp_reservation and raises on failure; lease deletion is best-effort/non-fatal since the Kea lease-delete endpoint shape is unverified against a real box. Co-Authored-By: Claude Sonnet 5 --- 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 df70ba7..bf02d3a 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1208,6 +1208,79 @@ class OPNsenseDriver(FirewallDriver): self._post("/api/kea/service/reconfigure") + def delete_dhcp_reservation_and_lease(self, mac: str, ip: str) -> dict[str, Any]: + """Remove a Kea DHCPv4 static reservation and any active lease for (mac, ip). + + Combined per NetOrk's VM-deletion cleanup flow — reservation and + lease removal are always requested together, so a single driver + call keeps callers from having to sequence two calls themselves. + + Reservation removal follows the same search-then-act pattern as + ``create_dhcp_reservation`` (``searchReservation`` -> ``del/{uuid}`` + -> ``service/reconfigure``) and raises on failure, mirroring that + method's "never silently no-op on the thing the caller explicitly + asked for" contract. + + Lease removal is best-effort and non-fatal: the exact Kea + lease-delete endpoint shape is unverified in this codebase (unlike + reservations, ``get_dhcp_leases()`` only ever implemented the read + path) — a failure here just means a stale lease record lingers in + Kea until its own natural cleanup, which is cosmetic, not a + functional problem (a deleted reservation already prevents the + client from getting the same IP back). + + :param mac: NIC MAC address of the reservation to remove. + :param ip: IP address of the reservation/lease to remove. + :returns: {"reservation_found", "reservation_deleted", "lease_found", + "lease_deleted"} — all bool. + :raises RuntimeError: Kea plugin unavailable, or Kea rejects + deletion of a reservation that does exist. + """ + result: dict[str, Any] = { + "reservation_found": False, + "reservation_deleted": False, + "lease_found": False, + "lease_deleted": False, + } + + # --- Reservation --- + try: + existing = self._post( + "/api/kea/dhcpv4/searchReservation", + {"current": 1, "rowCount": -1, "searchPhrase": ip}, + ) + except Exception as exc: + raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc + + reservation_uuid = None + for row in existing.get("rows") or []: + if row.get("ip_address") == ip: + reservation_uuid = row.get("uuid") + break + + if reservation_uuid: + result["reservation_found"] = True + del_result = self._post(f"/api/kea/dhcpv4/delReservation/{reservation_uuid}") + if del_result.get("result") != "deleted": + raise RuntimeError(f"Kea rejected reservation delete for {ip}: {del_result}") + result["reservation_deleted"] = True + self._post("/api/kea/service/reconfigure") + + # --- Lease (best-effort) --- + try: + leases = self._get("/api/kea/leases4/search") + rows = leases.get("rows") or leases.get("leases") or [] + if any((row.get("address") or row.get("ip-address")) == ip for row in rows): + result["lease_found"] = True + del_lease = self._post("/api/kea/leases4/delLease", {"ip-address": ip}) + result["lease_deleted"] = bool( + del_lease.get("result") == "deleted" or del_lease.get("status") == "ok" + ) + except Exception as exc: + logger.warning("DHCP lease delete for %s failed (non-fatal): %s", ip, exc) + + return result + 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 044b86a..fe40859 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1340,3 +1340,100 @@ class TestCreateDhcpReservation: driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253") driver.session.post.assert_not_called() + + +# --------------------------------------------------------------------------- +# delete_dhcp_reservation_and_lease() +# --------------------------------------------------------------------------- + + +class TestDeleteDhcpReservationAndLease: + def test_deletes_existing_reservation_and_lease(self, driver): + driver.session.post.side_effect = [ + _make_json_response( + {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]} + ), # searchReservation — match + _make_json_response({"result": "deleted"}), # delReservation + _make_json_response({"status": "ok"}), # service/reconfigure + _make_json_response({"result": "deleted"}), # delLease + ] + driver.session.get.return_value = _make_json_response( + {"rows": [{"address": "172.22.8.253", "hwaddr": "02:aa:bb:cc:dd:ee"}]} + ) + + result = driver.delete_dhcp_reservation_and_lease( + mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253" + ) + + assert result == { + "reservation_found": True, + "reservation_deleted": True, + "lease_found": True, + "lease_deleted": True, + } + del_res_call = driver.session.post.call_args_list[1] + assert ( + del_res_call.args[0] + == "https://opnsense.example.com/api/kea/dhcpv4/delReservation/existing-uuid-456" + ) + reconfigure_call = driver.session.post.call_args_list[2] + assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + del_lease_call = driver.session.post.call_args_list[3] + assert del_lease_call.args[0] == "https://opnsense.example.com/api/kea/leases4/delLease" + assert del_lease_call.kwargs["json"] == {"ip-address": "172.22.8.253"} + + def test_noop_when_no_reservation_and_no_lease_found(self, driver): + driver.session.post.return_value = _make_json_response({"rows": []}) + driver.session.get.return_value = _make_json_response({"rows": []}) + + result = driver.delete_dhcp_reservation_and_lease( + mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253" + ) + + assert result == { + "reservation_found": False, + "reservation_deleted": False, + "lease_found": False, + "lease_deleted": False, + } + # Only searchReservation was called — no delReservation/reconfigure/delLease. + assert driver.session.post.call_count == 1 + + def test_reservation_delete_raises_when_kea_rejects(self, driver): + driver.session.post.side_effect = [ + _make_json_response( + {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]} + ), + _make_json_response({"result": "not found"}), # delReservation rejected + ] + + with pytest.raises(RuntimeError, match="Kea rejected reservation delete"): + driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253") + + # reconfigure must NOT be called after a rejected delete + assert driver.session.post.call_count == 2 + + def test_lease_delete_failure_is_non_fatal(self, driver): + driver.session.post.side_effect = [ + _make_json_response( + {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]} + ), + _make_json_response({"result": "deleted"}), + _make_json_response({"status": "ok"}), + ] + driver.session.get.side_effect = Exception("connection reset") + + result = driver.delete_dhcp_reservation_and_lease( + mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253" + ) + + assert result["reservation_found"] is True + assert result["reservation_deleted"] is True + assert result["lease_found"] is False + assert result["lease_deleted"] is False + + def test_raises_when_kea_plugin_unavailable(self, driver): + driver.session.post.side_effect = Exception("404 Not Found") + + with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"): + driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")