From e51607a020754a35513a18ed31c8b2e072fc493f Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 15 Jul 2026 15:26:05 +0200 Subject: [PATCH] feat(freeradius): add NAS client and user CRUD driver methods get/create/delete_radius_client and get/create/delete_radius_user, backed by /api/freeradius/{client,user}/{search,add,del}_* and a reconfigure call to apply changes. Endpoints and field names (client.ip, not ipaddr) verified against a live OPNsense 24.7 instance via a real add -> search/get -> set -> del round trip, cleaned up immediately after. --- napalm_opnsense/opnsense.py | 83 ++++++++++++++++++ tests/unit/test_driver.py | 167 ++++++++++++++++++++++++++++++++++++ 2 files changed, 250 insertions(+) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index a4f25cc..a2c087f 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1118,6 +1118,89 @@ class OPNsenseDriver(FirewallDriver): "running": status.get("status") == "running", } + def get_radius_clients(self) -> list[dict[str, Any]]: + """Return configured FreeRADIUS NAS clients. + + Calls ``GET /api/freeradius/client/search_client``. + """ + try: + data = self._get("/api/freeradius/client/search_client") + except Exception as exc: + logger.warning("get_radius_clients() failed: %s", exc) + return [] + return [ + { + "id": row.get("uuid", ""), + "name": row.get("name", ""), + "ip": row.get("ip", ""), + "enabled": row.get("enabled") == "1", + } + for row in data.get("rows", []) + ] + + def create_radius_client(self, name: str, ip: str, secret: str) -> dict[str, Any]: + """Create a FreeRADIUS NAS client and apply the change. + + Calls ``POST /api/freeradius/client/add_client``, then + ``POST /api/freeradius/service/reconfigure`` to apply -- a saved + client has no effect on the running radiusd until reconfigured. + """ + payload = {"client": {"name": name, "ip": ip, "secret": secret}} + result = self._post("/api/freeradius/client/add_client", payload) + if result.get("result") != "saved": + return {"success": False, "validations": result.get("validations", {})} + self._post("/api/freeradius/service/reconfigure") + return {"success": True} + + def delete_radius_client(self, client_id: str) -> dict[str, Any]: + """Delete a FreeRADIUS NAS client by uuid and apply the change.""" + result = self._post(f"/api/freeradius/client/del_client/{client_id}") + if result.get("result") != "deleted": + return {"success": False} + self._post("/api/freeradius/service/reconfigure") + return {"success": True} + + def get_radius_users(self) -> list[dict[str, Any]]: + """Return configured FreeRADIUS users. + + Calls ``GET /api/freeradius/user/search_user``. Never includes the + password field. + """ + try: + data = self._get("/api/freeradius/user/search_user") + except Exception as exc: + logger.warning("get_radius_users() failed: %s", exc) + return [] + return [ + { + "id": row.get("uuid", ""), + "username": row.get("username", ""), + "enabled": row.get("enabled") == "1", + } + for row in data.get("rows", []) + ] + + def create_radius_user(self, username: str, password: str) -> dict[str, Any]: + """Create a FreeRADIUS user and apply the change. + + Calls ``POST /api/freeradius/user/add_user``, then + ``POST /api/freeradius/service/reconfigure`` to apply. + """ + payload = {"user": {"username": username, "password": password}} + result = self._post("/api/freeradius/user/add_user", payload) + if result.get("result") != "saved": + return {"success": False, "validations": result.get("validations", {})} + self._post("/api/freeradius/service/reconfigure") + return {"success": True} + + def delete_radius_user(self, user_id: str) -> dict[str, Any]: + """Delete a FreeRADIUS user by uuid and apply the change.""" + result = self._post(f"/api/freeradius/user/del_user/{user_id}") + if result.get("result") != "deleted": + return {"success": False} + self._post("/api/freeradius/service/reconfigure") + return {"success": True} + def get_dhcp_leases(self) -> list[dict[str, Any]]: """Return active DHCP leases from OPNsense. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 8d7f421..9373b7a 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1120,6 +1120,173 @@ class TestGetDdnsStatus: assert driver.get_ddns_status() is None +# --------------------------------------------------------------------------- +# FreeRADIUS: get/create/delete radius clients and users +# +# Endpoints + field names verified 2026-07-15 against a live OPNsense 24.7 +# instance via a real add -> search/get -> set -> del round trip (cleaned up +# immediately after). Key findings: the module is "freeradius" (matches the +# service id), the CRUD actions use a "search_"/"add_"/ +# "set_/"/"del_/" suffix convention (not the plain +# "search"/"add" used by e.g. get_certificates' Trust module), the client's +# IP field is literally named "ip" (CIDR string, e.g. "203.0.113.1/32") -- +# NOT "ipaddr" as first guessed (silently accepted-but-ignored by add_client +# rather than erroring). A config change only takes effect after +# POST /api/freeradius/service/reconfigure. +# --------------------------------------------------------------------------- + +RADIUS_CLIENTS_RESPONSE = { + "rows": [ + {"uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6", "enabled": "1", + "name": "ap-lobby", "secret": "s3cr3t", "ip": "10.0.0.5/32"}, + ] +} + +RADIUS_USERS_RESPONSE = { + "rows": [ + {"uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e", "enabled": "1", + "username": "jdoe", "password": "hunter2"}, + ] +} + + +class TestGetRadiusClients: + def test_returns_one_entry_per_row(self, driver): + driver._get = lambda path: RADIUS_CLIENTS_RESPONSE + assert len(driver.get_radius_clients()) == 1 + + def test_maps_expected_fields(self, driver): + driver._get = lambda path: RADIUS_CLIENTS_RESPONSE + client = driver.get_radius_clients()[0] + assert client["id"] == "e5483f1b-936b-47ba-8cca-24d56ad643c6" + assert client["name"] == "ap-lobby" + assert client["ip"] == "10.0.0.5/32" + assert client["enabled"] is True + + def test_calls_search_client(self, driver): + calls = [] + driver._get = lambda path: calls.append(path) or RADIUS_CLIENTS_RESPONSE + driver.get_radius_clients() + assert calls == ["/api/freeradius/client/search_client"] + + def test_api_error_returns_empty_list(self, driver): + driver._get = lambda path: (_ for _ in ()).throw(Exception("404")) + assert driver.get_radius_clients() == [] + + +class TestCreateRadiusClient: + def test_success_reconfigures_service(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + result = driver.create_radius_client("ap-lobby", "10.0.0.5/32", "s3cr3t") + assert result == {"success": True} + assert calls[0] == ( + "/api/freeradius/client/add_client", + {"client": {"name": "ap-lobby", "ip": "10.0.0.5/32", "secret": "s3cr3t"}}, + ) + assert calls[1] == ("/api/freeradius/service/reconfigure", None) + + def test_validation_failure_does_not_reconfigure(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or { + "result": "failed", + "validations": {"client.name": "A value is required."}, + } + result = driver.create_radius_client("", "10.0.0.5/32", "s3cr3t") + assert result["success"] is False + assert result["validations"] == {"client.name": "A value is required."} + assert calls == ["/api/freeradius/client/add_client"] + + +class TestDeleteRadiusClient: + def test_success_reconfigures_service(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or {"result": "deleted"} + result = driver.delete_radius_client("e5483f1b-936b-47ba-8cca-24d56ad643c6") + assert result == {"success": True} + assert calls == [ + "/api/freeradius/client/del_client/e5483f1b-936b-47ba-8cca-24d56ad643c6", + "/api/freeradius/service/reconfigure", + ] + + def test_not_found_does_not_reconfigure(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or {"result": "failed"} + result = driver.delete_radius_client("does-not-exist") + assert result == {"success": False} + assert calls == ["/api/freeradius/client/del_client/does-not-exist"] + + +class TestGetRadiusUsers: + def test_returns_one_entry_per_row(self, driver): + driver._get = lambda path: RADIUS_USERS_RESPONSE + assert len(driver.get_radius_users()) == 1 + + def test_maps_expected_fields(self, driver): + driver._get = lambda path: RADIUS_USERS_RESPONSE + user = driver.get_radius_users()[0] + assert user["id"] == "71da23fd-03f3-4e1f-a7e1-bb25645b981e" + assert user["username"] == "jdoe" + assert user["enabled"] is True + + def test_never_includes_password(self, driver): + driver._get = lambda path: RADIUS_USERS_RESPONSE + for user in driver.get_radius_users(): + assert "password" not in user + + def test_calls_search_user(self, driver): + calls = [] + driver._get = lambda path: calls.append(path) or RADIUS_USERS_RESPONSE + driver.get_radius_users() + assert calls == ["/api/freeradius/user/search_user"] + + def test_api_error_returns_empty_list(self, driver): + driver._get = lambda path: (_ for _ in ()).throw(Exception("404")) + assert driver.get_radius_users() == [] + + +class TestCreateRadiusUser: + def test_success_reconfigures_service(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + result = driver.create_radius_user("jdoe", "hunter2") + assert result == {"success": True} + assert calls[0] == ( + "/api/freeradius/user/add_user", + {"user": {"username": "jdoe", "password": "hunter2"}}, + ) + assert calls[1] == ("/api/freeradius/service/reconfigure", None) + + def test_validation_failure_does_not_reconfigure(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or { + "result": "failed", + "validations": {"user.username": "A value is required."}, + } + result = driver.create_radius_user("", "hunter2") + assert result["success"] is False + assert calls == ["/api/freeradius/user/add_user"] + + +class TestDeleteRadiusUser: + def test_success_reconfigures_service(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or {"result": "deleted"} + result = driver.delete_radius_user("71da23fd-03f3-4e1f-a7e1-bb25645b981e") + assert result == {"success": True} + assert calls == [ + "/api/freeradius/user/del_user/71da23fd-03f3-4e1f-a7e1-bb25645b981e", + "/api/freeradius/service/reconfigure", + ] + + def test_not_found_does_not_reconfigure(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or {"result": "failed"} + result = driver.delete_radius_user("does-not-exist") + assert result == {"success": False} + assert calls == ["/api/freeradius/user/del_user/does-not-exist"] + + # --------------------------------------------------------------------------- # _post() # ---------------------------------------------------------------------------