feat(freeradius): add NAS client and user CRUD driver methods
CI / test (3.10) (push) Failing after 7s
CI / test (3.11) (push) Failing after 7s
CI / test (3.12) (push) Failing after 8s
CI / test (3.9) (push) Failing after 7s

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.
This commit is contained in:
Christian Manivong
2026-07-15 15:26:05 +02:00
parent c8caa14176
commit e51607a020
2 changed files with 250 additions and 0 deletions
+83
View File
@@ -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.
+167
View File
@@ -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_<name>"/"add_<name>"/
# "set_<name>/<uuid>"/"del_<name>/<uuid>" 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()
# ---------------------------------------------------------------------------