Christian Manivong
8ba95a0709
Part of netork#85. Fills in the three device-specific methods the new DhcpServerMixin subnet layer expects. searchSubnet only carries uuid/subnet/description, so get_dhcp_subnets follows each row with getSubnet for the option data. That is one request per subnet; a firewall serves a handful, so the round trips cost less than the reconfigure they help avoid. An option Kea does not carry is omitted rather than reported as empty, because the generic diff reads an absent key as "not managed" — reporting [] would make every unmanaged option look like a pending change. apply_dhcp_subnet honours the mixin's partial-update contract: on an update it reads the subnet's current options first and replaces only the named ones. Without that, managing domain_search alone would blank the routers Kea autocollected and strand every client on that VLAN without a gateway. Setting any option also forces option_data_autocollect off — left on, Kea keeps re-filling routers/DNS/NTP and the next diff sees a change again, which is a reconfigure loop rather than a converged state. OPNsense renders repeatable option fields as comma-separated strings in some versions and as a selection map in others, for the same logical field. Both shapes are accepted rather than pinning the driver to one release. Pools are a newline-separated text block. A subnet whose detail fetch fails is skipped with a log line instead of aborting, same rule as get_dhcp_reservations: one broken record must not make the whole inventory unreadable. 16 new tests. Not yet verified against a live device — no reachable OPNsense at the time of writing, same caveat the reservation support shipped with.
2818 lines
119 KiB
Python
2818 lines
119 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||
# you may not use this file except in compliance with the License.
|
||
# You may obtain a copy of the License at
|
||
#
|
||
# http://www.apache.org/licenses/LICENSE-2.0
|
||
#
|
||
# Unless required by applicable law or agreed to in writing, software
|
||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
# See the License for the specific language governing permissions and
|
||
# limitations under the License.
|
||
|
||
"""NAPALM driver for OPNsense firewalls (REST API).
|
||
|
||
OPNsense exposes a JSON REST API at ``/api/``. Authentication uses an
|
||
API key/secret pair generated in the OPNsense GUI (System → Access →
|
||
Users → edit user → API keys). Pass them via *optional_args*:
|
||
|
||
optional_args={
|
||
"api_key": "<key>",
|
||
"api_secret": "<secret>",
|
||
"verify": False, # disable TLS verification for self-signed certs
|
||
}
|
||
|
||
If *api_key* / *api_secret* are omitted the driver falls back to the
|
||
positional *username* / *password* arguments.
|
||
|
||
Config management (load_merge_candidate / commit_config / rollback)
|
||
operates on **static routes** via ``/api/routes/routes/``. Other parts
|
||
of the OPNsense configuration (firewall rules, interface settings, …) are
|
||
not exposed through a generic config-push endpoint and must be managed
|
||
per-module through their respective API controllers.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import difflib
|
||
import json
|
||
import logging
|
||
import socket
|
||
from ipaddress import ip_address, ip_network
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
import requests
|
||
from requests.exceptions import RequestException
|
||
|
||
from napalm_device_types import FingerprintRule, FirewallDriver
|
||
from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException
|
||
|
||
from napalm_opnsense.ping_mixin import OPNsensePingMixin
|
||
|
||
|
||
class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
|
||
"""NAPALM driver for OPNsense (read-only, REST API)."""
|
||
|
||
VENDOR = "OPNsense"
|
||
DRIVER_NAME = "opnsense"
|
||
SNMP_FINGERPRINT = [
|
||
FingerprintRule("opnsense", weight=8.0, mandatory=True),
|
||
]
|
||
SSH_FINGERPRINT = [
|
||
FingerprintRule("freebsd", weight=7.0),
|
||
]
|
||
HTTP_FINGERPRINT = [
|
||
FingerprintRule("opnsense", weight=8.0, mandatory=True),
|
||
]
|
||
|
||
def __init__(
|
||
self,
|
||
hostname: str,
|
||
username: str,
|
||
password: str,
|
||
timeout: int = 60,
|
||
optional_args: dict[str, Any] | None = None,
|
||
) -> None:
|
||
self.hostname = hostname
|
||
self.username = username
|
||
self.password = password
|
||
self.timeout = timeout
|
||
self.optional_args = optional_args or {}
|
||
|
||
# NAPALM standard attributes
|
||
self.force_no_enable = True
|
||
self.use_canonical_interface = False
|
||
|
||
# OPNsense REST API settings
|
||
self.base_url = self.optional_args.get("base_url") or f"https://{hostname}"
|
||
# Accept "verify", "verify_ssl", or "ssl_verify" (NetOrk passes "verify_ssl")
|
||
_v = self.optional_args.get("verify", self.optional_args.get("verify_ssl", self.optional_args.get("ssl_verify", True)))
|
||
self.verify = bool(_v)
|
||
self.api_key = self.optional_args.get("api_key") or username
|
||
self.api_secret = self.optional_args.get("api_secret") or password
|
||
|
||
self.session: requests.Session | None = None
|
||
|
||
# Config-management state
|
||
self._candidate_config: list[dict[str, Any]] | None = None
|
||
# Backup ID of the config snapshot taken just before commit_config().
|
||
# Used by rollback() to restore the exact pre-commit state.
|
||
self._pre_commit_backup_id: str | None = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# Connection management
|
||
# ------------------------------------------------------------------
|
||
|
||
def open(self) -> None:
|
||
"""Open an HTTPS session to the OPNsense API and validate credentials."""
|
||
try:
|
||
s = requests.Session()
|
||
s.verify = self.verify
|
||
s.headers.update({"Accept": "application/json"})
|
||
s.auth = (self.api_key, self.api_secret)
|
||
self.session = s
|
||
# Lightweight connectivity and auth check
|
||
self._get("/api/core/system/status")
|
||
except RequestException as exc:
|
||
self.session = None
|
||
raise ConnectionException(
|
||
f"Cannot connect to OPNsense API at {self.base_url}: {exc}"
|
||
) from exc
|
||
|
||
def close(self) -> None:
|
||
"""Close the HTTPS session."""
|
||
if self.session is not None:
|
||
self.session.close()
|
||
self.session = None
|
||
|
||
def is_alive(self) -> dict[str, bool]:
|
||
"""Return whether the session is usable.
|
||
|
||
Performs a lightweight socket-level check without sending a full
|
||
HTTP request to avoid polluting API logs.
|
||
"""
|
||
if self.session is None:
|
||
return {"is_alive": False}
|
||
try:
|
||
host = self.hostname
|
||
port = 443
|
||
with socket.create_connection((host, port), timeout=5):
|
||
pass
|
||
return {"is_alive": True}
|
||
except (socket.error, OSError):
|
||
return {"is_alive": False}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Internal helpers
|
||
# ------------------------------------------------------------------
|
||
|
||
def _get(self, path: str) -> dict[str, Any]:
|
||
"""Perform a GET request against the OPNsense REST API.
|
||
|
||
:raises ConnectionClosedException: if called before :meth:`open`.
|
||
:raises RequestException: on HTTP-level errors.
|
||
"""
|
||
if self.session is None:
|
||
raise ConnectionClosedException("Not connected – call open() first.")
|
||
url = self.base_url.rstrip("/") + path
|
||
response = self.session.get(url, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
def _post(self, path: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""Perform a POST request against the OPNsense REST API.
|
||
|
||
:param path: API path, e.g. ``/api/routes/routes/addroute``.
|
||
:param data: JSON-serialisable payload (sent as ``application/json``).
|
||
:raises ConnectionClosedException: if called before :meth:`open`.
|
||
:raises RequestException: on HTTP-level errors.
|
||
"""
|
||
if self.session is None:
|
||
raise ConnectionClosedException("Not connected – call open() first.")
|
||
url = self.base_url.rstrip("/") + path
|
||
response = self.session.post(url, json=data or {}, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM getters
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_facts(self) -> dict[str, Any]:
|
||
"""Return a dictionary of general device facts.
|
||
|
||
Calls ``GET /api/core/system/status``.
|
||
|
||
Returned keys (NAPALM standard):
|
||
``vendor``, ``model``, ``hostname``, ``fqdn``, ``os_version``,
|
||
``serial_number``, ``uptime``, ``interface_list``.
|
||
"""
|
||
status = self._get("/api/core/system/status")
|
||
|
||
hostname = status.get("hostname") or status.get("name") or ""
|
||
version = status.get("version") or status.get("product_version") or "unknown"
|
||
|
||
try:
|
||
interface_list = list(self.get_interfaces().keys())
|
||
except Exception as exc:
|
||
logger.debug("Failed to fetch interface list: %s", exc)
|
||
interface_list = []
|
||
|
||
return {
|
||
"vendor": self.VENDOR,
|
||
"model": status.get("model") or self.VENDOR,
|
||
"hostname": hostname,
|
||
"fqdn": hostname,
|
||
"os_version": version,
|
||
"serial_number": status.get("serial") or "",
|
||
"uptime": status.get("uptime", -1),
|
||
"interface_list": interface_list,
|
||
}
|
||
|
||
def get_interfaces(self) -> dict[str, dict[str, Any]]:
|
||
"""Return interface details keyed by interface name.
|
||
|
||
Calls ``GET /api/interfaces/overview/export``.
|
||
|
||
Each entry contains NAPALM standard keys:
|
||
``is_up``, ``is_enabled``, ``description``, ``last_flapped``,
|
||
``mac_address``, ``speed``, ``mtu``.
|
||
"""
|
||
data = self._get("/api/interfaces/overview/export")
|
||
interfaces: dict[str, dict[str, Any]] = {}
|
||
|
||
# API returns a bare list in newer OPNsense versions;
|
||
# older/wrapped format uses {"interfaces": [...]}
|
||
items: list = data if isinstance(data, list) else data.get("interfaces", [])
|
||
|
||
for iface in items:
|
||
name = iface.get("device") or iface.get("name", "")
|
||
if not name:
|
||
continue
|
||
interfaces[name] = {
|
||
"is_up": iface.get("status", "") == "up" if "status" in iface else bool(iface.get("up", False)),
|
||
"is_enabled": bool(iface.get("enabled", True)),
|
||
"description": iface.get("description") or iface.get("descr") or "",
|
||
"last_flapped": -1.0,
|
||
"mac_address": (iface.get("macaddr") or iface.get("mac") or "").lower(),
|
||
"speed": float(iface["speed_mbps"]) if iface.get("speed_mbps") else 0.0,
|
||
"mtu": int(iface["mtu"]) if iface.get("mtu") else 0,
|
||
}
|
||
|
||
return interfaces
|
||
|
||
def get_interfaces_ip(self) -> dict[str, dict[str, Any]]:
|
||
"""Return IP addresses grouped by interface name.
|
||
|
||
Uses ``GET /api/interfaces/overview/export`` (same source as get_interfaces).
|
||
|
||
Structure::
|
||
|
||
{
|
||
"em0": {
|
||
"ipv4": {"192.0.2.10": {"prefix_length": 24}},
|
||
"ipv6": {"2001:db8::1": {"prefix_length": 64}},
|
||
}
|
||
}
|
||
"""
|
||
import ipaddress
|
||
|
||
data = self._get("/api/interfaces/overview/export")
|
||
items: list = data if isinstance(data, list) else data.get("interfaces", [])
|
||
result: dict[str, dict[str, Any]] = {}
|
||
|
||
def _add(ifname: str, cidr: str, family: str) -> None:
|
||
try:
|
||
iface_obj = ipaddress.ip_interface(cidr)
|
||
ip = str(iface_obj.ip)
|
||
prefix = iface_obj.network.prefixlen
|
||
result.setdefault(ifname, {"ipv4": {}, "ipv6": {}})
|
||
result[ifname][family][ip] = {"prefix_length": prefix}
|
||
except ValueError:
|
||
pass
|
||
|
||
for iface in items:
|
||
name: str = iface.get("device") or iface.get("name", "")
|
||
if not name:
|
||
continue
|
||
addr4: str = iface.get("addr4", "")
|
||
addr6: str = iface.get("addr6", "")
|
||
if addr4:
|
||
_add(name, addr4, "ipv4")
|
||
if addr6:
|
||
_add(name, addr6, "ipv6")
|
||
if not addr4:
|
||
for entry in iface.get("ipv4") or []:
|
||
ip_field = entry.get("ipaddr") or entry.get("ip", "")
|
||
subnet = entry.get("subnetbits") or entry.get("prefix_length")
|
||
cidr = f"{ip_field}/{subnet}" if subnet and "/" not in ip_field else ip_field
|
||
if cidr:
|
||
_add(name, cidr, "ipv4")
|
||
if not addr6:
|
||
for entry in iface.get("ipv6") or []:
|
||
ip_field = entry.get("ipaddr") or entry.get("ip", "")
|
||
prefix = entry.get("prefixlen") or entry.get("prefix_length")
|
||
cidr = f"{ip_field}/{prefix}" if prefix and "/" not in ip_field else ip_field
|
||
if cidr:
|
||
_add(name, cidr, "ipv6")
|
||
|
||
return result
|
||
|
||
def get_networks(self) -> list[dict[str, Any]]:
|
||
"""Return the IP networks this firewall is authoritative for.
|
||
|
||
Derived from the interface overview (``GET /api/interfaces/overview/export``).
|
||
Loopback and link-local addresses are excluded.
|
||
|
||
Each entry::
|
||
|
||
{
|
||
"network": "192.168.1.0/24",
|
||
"interface": "em0",
|
||
"gateway": "192.168.1.1",
|
||
"family": "ipv4",
|
||
"prefix_length": 24,
|
||
"vlan_id": 20,
|
||
}
|
||
|
||
``vlan_id`` is ``None`` for interfaces that are not 802.1Q VLAN
|
||
sub-interfaces (e.g. the untagged management/LAN interface).
|
||
"""
|
||
import ipaddress
|
||
|
||
data = self._get("/api/interfaces/overview/export")
|
||
items: list = data if isinstance(data, list) else data.get("interfaces", [])
|
||
|
||
networks: list[dict[str, Any]] = []
|
||
|
||
def _add(ifname: str, cidr: str, family: str, vlan_id: int | None) -> None:
|
||
"""Parse a CIDR string (e.g. '10.0.0.1/24') and append to networks."""
|
||
try:
|
||
iface_obj = ipaddress.ip_interface(cidr)
|
||
net = iface_obj.network
|
||
if net.is_loopback or net.is_link_local:
|
||
return
|
||
networks.append({
|
||
"network": str(net),
|
||
"interface": ifname,
|
||
"gateway": str(iface_obj.ip),
|
||
"family": family,
|
||
"prefix_length": net.prefixlen,
|
||
"vlan_id": vlan_id,
|
||
})
|
||
except ValueError:
|
||
pass
|
||
|
||
for iface in items:
|
||
name: str = iface.get("device") or iface.get("name", "")
|
||
if not name:
|
||
continue
|
||
|
||
vlan_id: int | None = None
|
||
raw_tag = iface.get("vlan_tag")
|
||
if raw_tag:
|
||
try:
|
||
vlan_id = int(raw_tag)
|
||
except (TypeError, ValueError):
|
||
vlan_id = None
|
||
|
||
# Primary format: flat CIDR strings in addr4/addr6
|
||
addr4: str = iface.get("addr4", "")
|
||
addr6: str = iface.get("addr6", "")
|
||
if addr4:
|
||
_add(name, addr4, "ipv4", vlan_id)
|
||
if addr6:
|
||
_add(name, addr6, "ipv6", vlan_id)
|
||
|
||
# Fallback: ipv4/ipv6 arrays where ipaddr may include prefix
|
||
if not addr4:
|
||
for entry in iface.get("ipv4") or []:
|
||
ip_field = entry.get("ipaddr") or entry.get("ip", "")
|
||
subnet = entry.get("subnetbits") or entry.get("prefix_length")
|
||
cidr = f"{ip_field}/{subnet}" if subnet and "/" not in ip_field else ip_field
|
||
if cidr:
|
||
_add(name, cidr, "ipv4", vlan_id)
|
||
if not addr6:
|
||
for entry in iface.get("ipv6") or []:
|
||
ip_field = entry.get("ipaddr") or entry.get("ip", "")
|
||
prefix = entry.get("prefixlen") or entry.get("prefix_length")
|
||
cidr = f"{ip_field}/{prefix}" if prefix and "/" not in ip_field else ip_field
|
||
if cidr:
|
||
_add(name, cidr, "ipv6", vlan_id)
|
||
|
||
return networks
|
||
|
||
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
|
||
"""Return the ARP table.
|
||
|
||
Calls ``GET /api/diagnostics/interface/get_arp``.
|
||
|
||
Each entry contains: ``interface``, ``mac``, ``ip``, ``age``.
|
||
"""
|
||
data = self._get("/api/diagnostics/interface/get_arp")
|
||
arp_table: list[dict[str, Any]] = []
|
||
|
||
for entry in data if isinstance(data, list) else data.get("arp", []):
|
||
arp_table.append(
|
||
{
|
||
"interface": entry.get("intf") or entry.get("interface", ""),
|
||
"mac": (entry.get("mac") or "").lower(),
|
||
"ip": entry.get("ip") or entry.get("address", ""),
|
||
"age": float(entry.get("expires", 0)),
|
||
}
|
||
)
|
||
|
||
return arp_table
|
||
|
||
def get_interfaces_counters(self) -> dict[str, dict[str, Any]]:
|
||
"""Return per-interface packet and byte counters.
|
||
|
||
Calls ``GET /api/diagnostics/interface/get_interface_statistics``.
|
||
|
||
Each entry contains NAPALM standard keys:
|
||
``tx_errors``, ``rx_errors``, ``tx_discards``, ``rx_discards``,
|
||
``tx_octets``, ``rx_octets``,
|
||
``tx_unicast_packets``, ``rx_unicast_packets``,
|
||
``tx_multicast_packets``, ``rx_multicast_packets``,
|
||
``tx_broadcast_packets``, ``rx_broadcast_packets``.
|
||
"""
|
||
data = self._get("/api/diagnostics/interface/get_interface_statistics")
|
||
counters: dict[str, dict[str, Any]] = {}
|
||
|
||
for iface, stats in data.get("statistics", {}).items():
|
||
counters[iface] = {
|
||
"tx_errors": int(stats.get("output-errors", 0)),
|
||
"rx_errors": int(stats.get("input-errors", 0)),
|
||
"tx_discards": int(stats.get("output-drops", 0)),
|
||
"rx_discards": int(stats.get("input-drops", 0)),
|
||
"tx_octets": int(stats.get("output-bytes", 0)),
|
||
"rx_octets": int(stats.get("input-bytes", 0)),
|
||
"tx_unicast_packets": int(stats.get("output-packets", 0)),
|
||
"rx_unicast_packets": int(stats.get("input-packets", 0)),
|
||
"tx_multicast_packets": int(stats.get("output-multicasts", 0)),
|
||
"rx_multicast_packets": int(stats.get("input-multicasts", 0)),
|
||
"tx_broadcast_packets": int(stats.get("output-broadcasts", 0)),
|
||
"rx_broadcast_packets": int(stats.get("input-broadcasts", 0)),
|
||
}
|
||
|
||
return counters
|
||
|
||
def get_environment(self) -> dict[str, Any]:
|
||
"""Return device environment data (CPU, memory, temperature).
|
||
|
||
Calls:
|
||
- ``GET /api/diagnostics/system/system_resources`` for CPU and memory.
|
||
- ``GET /api/diagnostics/system/system_temperature`` for temperature sensors.
|
||
|
||
``fan`` and ``power`` fields are not exposed by OPNsense and are
|
||
returned with assumed-healthy placeholder values.
|
||
"""
|
||
resources = self._get("/api/diagnostics/system/system_resources")
|
||
|
||
cpu_pct = float(resources.get("cpu", {}).get("used", 0))
|
||
mem_total = int(resources.get("memory", {}).get("total", 0) or 0)
|
||
mem_used = int(resources.get("memory", {}).get("used", 0) or 0)
|
||
|
||
try:
|
||
temp_data = self._get("/api/diagnostics/system/system_temperature")
|
||
except Exception as exc:
|
||
logger.debug("Failed to fetch system temperature: %s", exc)
|
||
temp_data = {}
|
||
|
||
temperature: dict[str, Any] = {}
|
||
for sensor in temp_data.get("data", []):
|
||
name = sensor.get("device") or sensor.get("name", "")
|
||
temp_val = float(sensor.get("temperature", 0))
|
||
temperature[name] = {
|
||
"temperature": temp_val,
|
||
"is_alert": temp_val > 80.0,
|
||
"is_critical": temp_val > 95.0,
|
||
}
|
||
|
||
return {
|
||
"fans": {},
|
||
"temperature": temperature,
|
||
"power": {},
|
||
"cpu": {0: {"%usage": cpu_pct}},
|
||
"memory": {
|
||
"available_ram": mem_total - mem_used,
|
||
"used_ram": mem_used,
|
||
},
|
||
}
|
||
|
||
def get_route_to(
|
||
self,
|
||
destination: str = "",
|
||
protocol: str = "",
|
||
longer: bool = False,
|
||
) -> dict[str, list[dict[str, Any]]]:
|
||
"""Return routing table entries.
|
||
|
||
Calls ``GET /api/diagnostics/interface/get_routes``.
|
||
|
||
:param destination: Filter by exact prefix (e.g. ``"192.0.2.0/24"``).
|
||
:param protocol: Filter by protocol name (``"static"``, ``"connected"``).
|
||
:param longer: Ignored (OPNsense does not support longer-prefixes filter).
|
||
|
||
Returns a NAPALM-standard route dict keyed by network prefix.
|
||
"""
|
||
data = self._get("/api/diagnostics/interface/get_routes")
|
||
routes: dict[str, list[dict[str, Any]]] = {}
|
||
|
||
# Optionally enrich with OSPF routes from FRR/Quagga
|
||
ospf_networks: set = set()
|
||
try:
|
||
ospf_data = self._get("/api/quagga/ospf/routes")
|
||
for prefix in (ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys()):
|
||
ospf_networks.add(str(prefix))
|
||
except Exception as exc:
|
||
logger.debug("Failed to fetch OSPF routes: %s", exc)
|
||
|
||
route_list = data if isinstance(data, list) else data.get("route", [])
|
||
for route in route_list:
|
||
network = route.get("network") or route.get("destination", "")
|
||
if not network:
|
||
continue
|
||
|
||
if destination and network != destination:
|
||
continue
|
||
|
||
flags = route.get("flags", "").upper()
|
||
gateway = route.get("gateway") or route.get("nexthop", "")
|
||
iface = route.get("netif") or route.get("interface", "")
|
||
|
||
# Clean up BSD link-layer gateway references
|
||
clean_gateway = "" if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") else gateway
|
||
|
||
# Determine address family from network address
|
||
family = "ipv6" if (":" in network or (gateway and ":" in gateway)) else "ipv4"
|
||
|
||
# Determine routing protocol from BSD flags:
|
||
# S = Static, dynamic routes have no S flag
|
||
if network in ospf_networks:
|
||
proto = "ospf"
|
||
elif "S" in flags:
|
||
proto = "static"
|
||
elif not clean_gateway:
|
||
proto = "connected"
|
||
else:
|
||
proto = "kernel"
|
||
|
||
if protocol and proto != protocol.lower():
|
||
continue
|
||
|
||
entry: dict[str, Any] = {
|
||
"protocol": proto,
|
||
"family": family,
|
||
"current_active": "U" in flags,
|
||
"last_active": False,
|
||
"age": -1,
|
||
"next_hop": clean_gateway,
|
||
"outgoing_interface": iface,
|
||
"selected_next_hop": True,
|
||
"preference": int(route.get("priority", 0)),
|
||
"inactive_reason": "",
|
||
"routing_table": "global",
|
||
"protocol_attributes": {},
|
||
}
|
||
routes.setdefault(network, []).append(entry)
|
||
|
||
return routes
|
||
|
||
def get_ipv6_neighbors_table(self) -> list[dict[str, Any]]:
|
||
"""Return the IPv6 Neighbor Discovery (NDP) table.
|
||
|
||
Calls ``GET /api/diagnostics/interface/get_ndp``.
|
||
|
||
Each entry contains: ``interface``, ``mac``, ``ip``, ``age``,
|
||
``state`` (best-effort from NDP flags).
|
||
"""
|
||
data = self._get("/api/diagnostics/interface/get_ndp")
|
||
neighbors: list[dict[str, Any]] = []
|
||
|
||
rows = data if isinstance(data, list) else data.get("rows", [])
|
||
for entry in rows:
|
||
neighbors.append(
|
||
{
|
||
"interface": entry.get("intf") or entry.get("interface", ""),
|
||
"mac": (entry.get("mac") or "").lower(),
|
||
"ip": entry.get("ip") or entry.get("address", ""),
|
||
"age": float(entry.get("expires", 0)),
|
||
"state": entry.get("state", ""),
|
||
}
|
||
)
|
||
|
||
return neighbors
|
||
|
||
def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
|
||
"""Return LLDP neighbors grouped by local port.
|
||
|
||
Calls ``GET /api/lldpd/service/neighbor``.
|
||
|
||
Requires the ``os-lldpd`` plugin to be installed on OPNsense.
|
||
Returns an empty dict if the plugin is not present.
|
||
"""
|
||
try:
|
||
data = self._get("/api/lldpd/service/neighbor")
|
||
except Exception as exc:
|
||
logger.debug("LLDP neighbor fetch failed (plugin not installed?): %s", exc)
|
||
return {}
|
||
|
||
neighbors: dict[str, list[dict[str, Any]]] = {}
|
||
for row in data.get("rows", []):
|
||
port = row.get("local_port") or row.get("port", "")
|
||
neighbors.setdefault(port, []).append(
|
||
{
|
||
"hostname": row.get("system_name") or row.get("chassis", ""),
|
||
"port": row.get("port_id") or row.get("port", ""),
|
||
}
|
||
)
|
||
return neighbors
|
||
|
||
def get_lldp_neighbors_detail(
|
||
self, interface: str = ""
|
||
) -> dict[str, list[dict[str, Any]]]:
|
||
"""Return detailed LLDP neighbor information.
|
||
|
||
Calls ``GET /api/lldpd/service/neighbor``.
|
||
|
||
Requires the ``os-lldpd`` plugin. Returns an empty dict if the
|
||
plugin is not available.
|
||
|
||
:param interface: If set, filter results to this local port.
|
||
"""
|
||
try:
|
||
data = self._get("/api/lldpd/service/neighbor")
|
||
except Exception as exc:
|
||
logger.debug("LLDP neighbor detail fetch failed (plugin not installed?): %s", exc)
|
||
return {}
|
||
|
||
details: dict[str, list[dict[str, Any]]] = {}
|
||
for row in data.get("rows", []):
|
||
port = row.get("local_port") or row.get("port", "")
|
||
if interface and port != interface:
|
||
continue
|
||
details.setdefault(port, []).append(
|
||
{
|
||
"parent_interface": "",
|
||
"remote_port": row.get("port_id") or row.get("port", ""),
|
||
"remote_port_description": row.get("port_description", ""),
|
||
"remote_chassis_id": row.get("chassis_id") or row.get("chassis", ""),
|
||
"remote_system_name": row.get("system_name", ""),
|
||
"remote_system_description": row.get("system_description", ""),
|
||
"remote_system_capab": [
|
||
c.strip().lower()
|
||
for c in row.get("system_capabilities", "").split(",")
|
||
if c.strip()
|
||
],
|
||
"remote_system_enable_capab": [
|
||
c.strip().lower()
|
||
for c in row.get("enabled_capabilities", "").split(",")
|
||
if c.strip()
|
||
],
|
||
}
|
||
)
|
||
return details
|
||
|
||
def get_ntp_servers(self) -> dict[str, dict[str, Any]]:
|
||
"""Return configured NTP servers.
|
||
|
||
Calls ``GET /api/ntpd/service/status`` which includes the list of
|
||
configured peer addresses in the ``peers`` field.
|
||
|
||
Returns a dict keyed by server address with an empty value dict
|
||
(NAPALM standard format).
|
||
"""
|
||
try:
|
||
data = self._get("/api/ntpd/service/status")
|
||
except Exception as exc:
|
||
logger.debug("NTP status fetch failed (plugin not installed?): %s", exc)
|
||
return {}
|
||
|
||
servers: dict[str, dict[str, Any]] = {}
|
||
for peer in data.get("peers", []):
|
||
addr = peer.get("address") or peer.get("remote", "")
|
||
if addr:
|
||
servers[addr] = {}
|
||
return servers
|
||
|
||
def get_vlans(self) -> dict[str, dict[str, Any]]:
|
||
"""Return configured VLANs.
|
||
|
||
Calls ``GET /api/interfaces/vlan_settings/search_item``.
|
||
|
||
Each VLAN device configured under *Interfaces → Other Types → VLAN*
|
||
becomes one entry, keyed by VLAN tag (as a string).
|
||
|
||
The ``interfaces`` list contains the VLAN device name (e.g.
|
||
``em0_vlan10``). When a device is already assigned to a logical
|
||
interface OPNsense appends the logical name in brackets
|
||
(``"em0_vlan10 [LAN]"``); this driver strips that annotation and
|
||
stores the bare device name.
|
||
|
||
Returned keys (NAPALM standard):
|
||
``name`` (description, or device name if no description is set),
|
||
``interfaces`` (list with the VLAN device name).
|
||
"""
|
||
data = self._get("/api/interfaces/vlan_settings/search_item")
|
||
vlans: dict[str, dict[str, Any]] = {}
|
||
|
||
for row in data.get("rows", []):
|
||
tag = str(row.get("tag", "")).strip()
|
||
if not tag:
|
||
continue
|
||
|
||
# vlanif may be "em0_vlan10 [LAN]" when assigned to a logical interface
|
||
vlanif_raw = str(row.get("vlanif", ""))
|
||
vlanif = vlanif_raw.split(" [")[0].strip()
|
||
|
||
descr = str(row.get("descr", "")).strip()
|
||
vlans[tag] = {
|
||
"name": descr or vlanif,
|
||
"interfaces": [vlanif] if vlanif else [],
|
||
}
|
||
|
||
return vlans
|
||
|
||
def get_bgp_neighbors(self) -> dict[str, Any]:
|
||
"""Return BGP neighbor state.
|
||
|
||
Requires the FRR plugin (``os-frr``) to be installed on OPNsense.
|
||
Returns an empty dict if the plugin is absent or FRR is not running.
|
||
|
||
Calls:
|
||
|
||
- ``GET /api/quagga/bgp/get`` — local AS number and router-id from
|
||
the BGP configuration model.
|
||
- ``GET /api/quagga/diagnostics/bgpneighbors`` — live neighbor state
|
||
as returned by FRR (``vtysh -c "show bgp neighbors json"``).
|
||
|
||
Returns a NAPALM-standard dict keyed by VRF name. Only the default
|
||
VRF (``"global"``) is populated; per-VRF BGP is not yet mapped.
|
||
|
||
Each peer entry contains:
|
||
``local_as``, ``remote_as``, ``remote_id``, ``is_up``,
|
||
``is_enabled``, ``description``, ``uptime``,
|
||
``address_family`` (``ipv4`` and/or ``ipv6`` with prefix counters).
|
||
"""
|
||
try:
|
||
bgp_cfg = self._get("/api/quagga/bgp/get")
|
||
neighbors_data = self._get("/api/quagga/diagnostics/bgpneighbors")
|
||
except Exception as exc:
|
||
logger.debug("BGP config fetch failed (plugin not installed?): %s", exc)
|
||
return {}
|
||
|
||
bgp = bgp_cfg.get("bgp", {})
|
||
local_as_default = int(bgp.get("asnumber", 0) or 0)
|
||
router_id = bgp.get("routerid", "")
|
||
|
||
raw_neighbors = neighbors_data.get("response", {})
|
||
if not isinstance(raw_neighbors, dict):
|
||
return {}
|
||
|
||
# FRR address-family key → NAPALM address-family key
|
||
_AF_MAP = {
|
||
"ipv4Unicast": "ipv4",
|
||
"ipv6Unicast": "ipv6",
|
||
}
|
||
|
||
peers: dict[str, Any] = {}
|
||
for peer_ip, nbr in raw_neighbors.items():
|
||
if not isinstance(nbr, dict):
|
||
continue
|
||
|
||
is_up = nbr.get("bgpState", "").lower() == "established"
|
||
uptime_msec = int(nbr.get("bgpTimerUpMsec", 0) or 0)
|
||
uptime = uptime_msec // 1000 if is_up else -1
|
||
|
||
address_family: dict[str, Any] = {}
|
||
for frr_af, napalm_af in _AF_MAP.items():
|
||
af = nbr.get("addressFamilyInfo", {}).get(frr_af)
|
||
if af is not None:
|
||
address_family[napalm_af] = {
|
||
"sent_prefixes": int(af.get("sentPrefixCounter", 0) or 0),
|
||
"received_prefixes": int(af.get("prefixReceivedCount", 0) or 0),
|
||
"accepted_prefixes": int(af.get("acceptedPrefixCounter", 0) or 0),
|
||
}
|
||
|
||
if not address_family:
|
||
# FRR did not report AF info (session not yet established)
|
||
address_family["ipv4"] = {
|
||
"sent_prefixes": -1,
|
||
"received_prefixes": -1,
|
||
"accepted_prefixes": -1,
|
||
}
|
||
|
||
peers[peer_ip] = {
|
||
"local_as": int(nbr.get("localAs", local_as_default) or local_as_default),
|
||
"remote_as": int(nbr.get("remoteAs", 0) or 0),
|
||
"remote_id": nbr.get("remoteRouterId", ""),
|
||
"is_up": is_up,
|
||
"is_enabled": not bool(nbr.get("adminShutdown", False)),
|
||
"description": nbr.get("nbrDesc", ""),
|
||
"uptime": uptime,
|
||
"address_family": address_family,
|
||
}
|
||
|
||
return {
|
||
"global": {
|
||
"router_id": router_id,
|
||
"peers": peers,
|
||
}
|
||
}
|
||
|
||
def get_config(
|
||
self,
|
||
retrieve: str = "all",
|
||
full: bool = False,
|
||
sanitized: bool = False,
|
||
format: str = "text",
|
||
) -> dict[str, str]:
|
||
"""Return device configuration.
|
||
|
||
OPNsense stores its configuration as XML. This getter returns the
|
||
raw XML text in the ``running`` slot. ``startup`` mirrors ``running``
|
||
(OPNsense applies config immediately). The ``candidate`` slot shows
|
||
the JSON-serialised staged routes when a candidate has been loaded,
|
||
or an empty string otherwise.
|
||
|
||
Calls ``GET /api/core/backup/download/this``.
|
||
"""
|
||
configs: dict[str, str] = {"running": "", "startup": "", "candidate": ""}
|
||
|
||
if retrieve in ("all", "running", "startup"):
|
||
try:
|
||
response = self._get("/api/core/backup/download/this")
|
||
xml_text: str = (
|
||
response if isinstance(response, str) else str(response)
|
||
)
|
||
if retrieve in ("all", "running"):
|
||
configs["running"] = xml_text
|
||
if retrieve in ("all", "startup"):
|
||
configs["startup"] = xml_text
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch running/startup config: %s", exc)
|
||
|
||
if retrieve in ("all", "candidate") and self._candidate_config is not None:
|
||
configs["candidate"] = json.dumps(self._candidate_config, indent=2)
|
||
|
||
return configs
|
||
|
||
# ------------------------------------------------------------------
|
||
# Config management (static routes)
|
||
# ------------------------------------------------------------------
|
||
|
||
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
|
||
"""Stage a set of static-route additions as a candidate config.
|
||
|
||
OPNsense does not offer a single generic config-push endpoint.
|
||
This method targets the **routes** subsystem
|
||
(``/api/routes/routes/``) and accepts a JSON list of route objects.
|
||
|
||
The *config* parameter must be a JSON string containing a list of
|
||
route dicts. Each dict may contain the following keys:
|
||
|
||
.. code-block:: json
|
||
|
||
[
|
||
{
|
||
"network": "10.0.0.0/8",
|
||
"gateway": "WAN_GW",
|
||
"descr": "optional description",
|
||
"disabled": "0"
|
||
}
|
||
]
|
||
|
||
``gateway`` must be the **name** of an existing OPNsense gateway
|
||
(not an IP address) as shown in
|
||
*System → Gateways → Configuration*.
|
||
|
||
:param filename: Path to a JSON file containing the route list.
|
||
:param config: JSON string containing the route list.
|
||
:raises MergeConfigException: if neither or both arguments are given,
|
||
or if the JSON is malformed.
|
||
"""
|
||
if filename is None and config is None:
|
||
raise MergeConfigException("Provide either 'filename' or 'config'.")
|
||
if filename is not None and config is not None:
|
||
raise MergeConfigException("Provide either 'filename' or 'config', not both.")
|
||
|
||
if filename is not None:
|
||
try:
|
||
with open(filename, "r", encoding="utf-8") as fh:
|
||
config = fh.read()
|
||
except OSError as exc:
|
||
raise MergeConfigException(f"Cannot read file {filename!r}: {exc}") from exc
|
||
|
||
try:
|
||
routes = json.loads(config) # type: ignore[arg-type]
|
||
except json.JSONDecodeError as exc:
|
||
raise MergeConfigException(f"Invalid JSON in candidate config: {exc}") from exc
|
||
|
||
if not isinstance(routes, list):
|
||
raise MergeConfigException(
|
||
"Candidate config must be a JSON array of route objects."
|
||
)
|
||
for i, route in enumerate(routes):
|
||
if not isinstance(route, dict):
|
||
raise MergeConfigException(f"Route at index {i} must be a JSON object.")
|
||
if "network" not in route or "gateway" not in route:
|
||
raise MergeConfigException(
|
||
f"Route at index {i} is missing 'network' or 'gateway'."
|
||
)
|
||
|
||
self._candidate_config = routes
|
||
|
||
def compare_config(self) -> str:
|
||
"""Return a unified diff of the candidate vs the current routes.
|
||
|
||
Fetches the live route list from ``GET /api/routes/routes/searchroute``
|
||
and diffs it against the staged candidate.
|
||
|
||
Returns an empty string if no candidate has been loaded.
|
||
"""
|
||
if self._candidate_config is None:
|
||
return ""
|
||
|
||
current_routes = self._fetch_current_routes()
|
||
current_text = json.dumps(current_routes, indent=2, sort_keys=True)
|
||
candidate_text = json.dumps(self._candidate_config, indent=2, sort_keys=True)
|
||
|
||
diff = difflib.unified_diff(
|
||
current_text.splitlines(keepends=True),
|
||
candidate_text.splitlines(keepends=True),
|
||
fromfile="current",
|
||
tofile="candidate",
|
||
)
|
||
return "".join(diff)
|
||
|
||
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
|
||
"""Apply the staged candidate routes to the device.
|
||
|
||
Each route in the candidate is submitted via
|
||
``POST /api/routes/routes/addroute``. After all routes are added,
|
||
``POST /api/routes/routes/reconfigure`` is called to activate them.
|
||
|
||
The UUIDs returned by OPNsense are stored internally so that
|
||
:meth:`rollback` can remove exactly these routes.
|
||
|
||
:param message: Ignored (OPNsense has no commit-message concept).
|
||
:param revert_in: Ignored (auto-rollback not supported via API).
|
||
:raises MergeConfigException: if no candidate is staged, or if the
|
||
API returns an error for any route.
|
||
"""
|
||
if self._candidate_config is None:
|
||
raise MergeConfigException("No candidate config loaded. Call load_merge_candidate() first.")
|
||
|
||
# Record the current backup ID so rollback() can restore exactly
|
||
# this state after OPNsense writes the new config to disk.
|
||
self._pre_commit_backup_id = self._get_latest_backup_id()
|
||
|
||
try:
|
||
for route in self._candidate_config:
|
||
payload = {
|
||
"route": {
|
||
"network": route["network"],
|
||
"gateway": route["gateway"],
|
||
"descr": route.get("descr", ""),
|
||
"disabled": route.get("disabled", "0"),
|
||
}
|
||
}
|
||
self._post("/api/routes/routes/addroute", payload)
|
||
|
||
self._post("/api/routes/routes/reconfigure")
|
||
except RequestException as exc:
|
||
raise MergeConfigException(f"Failed to apply route config: {exc}") from exc
|
||
|
||
self._candidate_config = None
|
||
|
||
def discard_config(self) -> None:
|
||
"""Discard the staged candidate config without applying it."""
|
||
self._candidate_config = None
|
||
|
||
def rollback(self) -> None:
|
||
"""Revert the device to the configuration state captured before the last commit.
|
||
|
||
OPNsense automatically saves a backup of ``config.xml`` before applying
|
||
configuration changes. :meth:`commit_config` records the ID of the
|
||
most recent backup at the time of the commit so that this method can
|
||
restore exactly the pre-commit state via
|
||
``POST /api/core/backup/revert_backup/{backup_id}``.
|
||
|
||
If no backup ID is available (i.e. :meth:`commit_config` was never
|
||
called in this session, or the backup list was empty at commit time),
|
||
the most recent backup from the device is used as a fallback.
|
||
|
||
If no backups exist at all, this is a no-op.
|
||
"""
|
||
backup_id = self._pre_commit_backup_id or self._get_latest_backup_id()
|
||
if backup_id is None:
|
||
return
|
||
|
||
self._post(f"/api/core/backup/revert_backup/{backup_id}")
|
||
self._pre_commit_backup_id = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# Internal helpers (config management)
|
||
# ------------------------------------------------------------------
|
||
|
||
def _get_latest_backup_id(self) -> str | None:
|
||
"""Return the ID of the most recent server-side config backup, or ``None``.
|
||
|
||
Calls ``GET /api/core/backup/backups/this``. The response is sorted
|
||
newest-first by OPNsense. Returns ``None`` when no backups exist or
|
||
the request fails.
|
||
"""
|
||
try:
|
||
data = self._get("/api/core/backup/backups/this")
|
||
items = data.get("items", [])
|
||
return items[0]["id"] if items else None
|
||
except Exception as exc:
|
||
logger.debug("No backup snapshots found: %s", exc)
|
||
return None
|
||
|
||
def _fetch_current_routes(self) -> list[dict[str, Any]]:
|
||
"""Return the current static routes from the OPNsense API.
|
||
|
||
Calls ``GET /api/routes/routes/searchroute`` and normalises the
|
||
response to the same keys used by :meth:`load_merge_candidate`.
|
||
"""
|
||
data = self._get("/api/routes/routes/searchroute")
|
||
routes: list[dict[str, Any]] = []
|
||
for row in data.get("rows", []):
|
||
routes.append(
|
||
{
|
||
"network": row.get("network", ""),
|
||
"gateway": row.get("gateway", ""),
|
||
"descr": row.get("descr", ""),
|
||
"disabled": row.get("disabled", "0"),
|
||
}
|
||
)
|
||
return routes
|
||
|
||
# ------------------------------------------------------------------
|
||
# NetOrch extensions: packages, services, updates
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_packages(self) -> list[dict[str, Any]]:
|
||
"""Return installed OPNsense plugins.
|
||
|
||
Calls ``GET /api/core/firmware/info`` and returns the ``plugin``
|
||
list filtered to entries where ``installed == "1"``. The format
|
||
mirrors the OpenWrt NAPALM driver so NetOrch can render both
|
||
drivers with the same UI component:
|
||
``{name, version, installed, description, size, source}``
|
||
"""
|
||
info = self._get("/api/core/firmware/info")
|
||
result: list[dict[str, Any]] = []
|
||
for p in info.get("plugin", []):
|
||
if p.get("installed") != "1":
|
||
continue
|
||
result.append({
|
||
"name": p.get("name", ""),
|
||
"version": p.get("version", ""),
|
||
"installed": True,
|
||
"description": p.get("comment", ""),
|
||
"size": 0,
|
||
"source": "opnsense-plugins",
|
||
})
|
||
return sorted(result, key=lambda x: x["name"].lower())
|
||
|
||
def get_certificates(self) -> list[dict[str, Any]]:
|
||
"""Return certificates from the OPNsense Trust store.
|
||
|
||
Calls ``POST /api/trust/cert/search`` and normalises each row to
|
||
``{name, issuer, valid_from, valid_to, in_use_by}`` (Unix
|
||
timestamps for the validity fields). Never includes
|
||
crt_payload/prv_payload/csr_payload -- those rows carry private key
|
||
material and must not leave the Trust store.
|
||
"""
|
||
try:
|
||
data = self._post("/api/trust/cert/search")
|
||
except Exception as exc:
|
||
logger.warning("get_certificates() failed: %s", exc)
|
||
return []
|
||
|
||
def _as_int(value: Any) -> int:
|
||
try:
|
||
return int(value or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for row in data.get("rows", []):
|
||
result.append({
|
||
"name": row.get("commonname") or row.get("descr") or row.get("refid", ""),
|
||
"issuer": row.get("%caref") or "",
|
||
"valid_from": _as_int(row.get("valid_from")),
|
||
"valid_to": _as_int(row.get("valid_to")),
|
||
"in_use_by": _as_int(row.get("in_use")),
|
||
})
|
||
return result
|
||
|
||
def get_ddns_status(self) -> dict[str, Any] | None:
|
||
"""Return Dynamic DNS (os-ddclient) enabled/running state.
|
||
|
||
Calls ``GET /api/dyndns/service/status`` and
|
||
``GET /api/dyndns/settings/get`` -- note the API module is
|
||
"dyndns", not "ddclient" (the service id reported by
|
||
get_services() is "ddclient", but its REST controller lives under
|
||
a different, historical module name).
|
||
|
||
Returns ``{"enabled": bool, "running": bool}``, or ``None`` if the
|
||
plugin isn't installed/reachable. Scoped to enabled/running only:
|
||
no per-account "registered IP" comparison, since that would need a
|
||
configured account to verify the response shape against.
|
||
"""
|
||
try:
|
||
status = self._get("/api/dyndns/service/status")
|
||
settings = self._get("/api/dyndns/settings/get")
|
||
except Exception as exc:
|
||
logger.warning("get_ddns_status() failed: %s", exc)
|
||
return None
|
||
|
||
general = settings.get("ddclient", {}).get("general", {})
|
||
return {
|
||
"enabled": general.get("enabled") == "1",
|
||
"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.
|
||
The add response carries no id, so the new client's uuid is looked
|
||
up afterwards via get_radius_clients() (matched by name) -- callers
|
||
need it to address this client in later set_client/del_client calls.
|
||
"""
|
||
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")
|
||
created = next((c for c in self.get_radius_clients() if c["name"] == name), None)
|
||
return {"success": True, "id": created["id"] if created else None}
|
||
|
||
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. The add
|
||
response carries no id, so the new user's uuid is looked up
|
||
afterwards via get_radius_users() (matched by username) -- callers
|
||
need it to address this user in later set_user/del_user calls.
|
||
"""
|
||
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")
|
||
created = next((u for u in self.get_radius_users() if u["username"] == username), None)
|
||
return {"success": True, "id": created["id"] if created else None}
|
||
|
||
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.
|
||
|
||
Tries all known DHCP backends in order:
|
||
|
||
1. **Kea DHCPv4** (os-kea plugin) — ``GET /api/kea/leases4/search``
|
||
Fields: ``hw-address``, ``ip-address``, ``hostname``, ``expire``
|
||
2. **ISC DHCP** (legacy) — ``POST /api/dhcpv4/leases/searchlease``
|
||
Fields: ``mac``, ``address``, ``hostname``, ``ends``
|
||
3. **ARP table** fallback — ``GET /api/diagnostics/interface/get_arp``
|
||
Provides IP only, hostname will be empty.
|
||
|
||
Each returned entry contains:
|
||
|
||
* ``mac`` — lower-case MAC address
|
||
* ``ip`` — assigned IP address
|
||
* ``hostname`` — client hostname (may be empty)
|
||
* ``lease_end`` — Unix timestamp when the lease expires (0 if unknown)
|
||
* ``state`` — raw state string
|
||
"""
|
||
def _parse_kea(rows: list[dict]) -> list[dict[str, Any]]:
|
||
result = []
|
||
for row in rows:
|
||
# OPNsense Kea uses "hwaddr"; standard Kea uses "hw-address"; ISC DHCP uses "mac"
|
||
mac = (
|
||
row.get("hwaddr") or row.get("hw-address") or row.get("mac") or ""
|
||
).lower().strip()
|
||
if not mac:
|
||
continue
|
||
raw_end = row.get("expire") or row.get("ends") or 0
|
||
try:
|
||
lease_end = int(raw_end)
|
||
except (ValueError, TypeError):
|
||
lease_end = 0
|
||
result.append({
|
||
"mac": mac,
|
||
"ip": (row.get("address") or row.get("ip-address") or row.get("ip") or "").strip(),
|
||
"hostname": (row.get("hostname") or "").strip(),
|
||
"lease_end": lease_end,
|
||
"state": str(row.get("state") or ""),
|
||
})
|
||
return result
|
||
|
||
# 1. Kea DHCPv4 plugin
|
||
try:
|
||
data = self._get("/api/kea/leases4/search")
|
||
rows = data.get("rows") or data.get("leases") or (data if isinstance(data, list) else [])
|
||
leases = _parse_kea(rows)
|
||
if leases:
|
||
return leases
|
||
except Exception as exc:
|
||
logger.debug("Kea DHCP lease fetch failed: %s", exc)
|
||
|
||
# 2. ISC DHCP (legacy)
|
||
try:
|
||
data = self._post(
|
||
"/api/dhcpv4/leases/searchlease",
|
||
{"current": 1, "rowCount": -1, "searchPhrase": "", "sort": {}},
|
||
)
|
||
leases = _parse_kea(data.get("rows", []))
|
||
if leases:
|
||
return leases
|
||
except Exception as exc:
|
||
logger.debug("ISC DHCP lease fetch failed: %s", exc)
|
||
|
||
# 3. ARP table fallback (IP only, no hostname)
|
||
try:
|
||
arp = self.get_arp_table()
|
||
return [
|
||
{"mac": e["mac"], "ip": e["ip"], "hostname": "", "lease_end": 0, "state": "arp"}
|
||
for e in arp
|
||
if e.get("mac") and e.get("ip")
|
||
]
|
||
except Exception as exc:
|
||
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 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<X>/{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 — 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). Endpoint verified live against a real OPNsense instance:
|
||
``LeasesController`` is documented as "Abstract [non-callable]" with
|
||
a ``del_lease($ips=null)`` action
|
||
(https://docs.opnsense.org/development/api/core/kea.html) — the
|
||
concrete, callable route is the ``leases4`` controller (matching
|
||
``search``, used above), and despite the ``$ips`` parameter name the
|
||
IP is passed as a URL path segment, not a JSON body field — a POST
|
||
body of ``{"ips": [ip]}`` (the natural reading of the signature)
|
||
returns ``{"status": "error", "message": "Missing lease IP
|
||
parameter"}``; only ``POST /api/kea/leases4/del_lease/{ip}`` works.
|
||
|
||
: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(f"/api/kea/leases4/del_lease/{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
|
||
|
||
# ------------------------------------------------------------------
|
||
# DhcpServerMixin implementation (Kea DHCPv4). The diff and the apply
|
||
# loop are generic and live in napalm_device_types.dhcp; only these
|
||
# three methods know about Kea's REST shape.
|
||
# ------------------------------------------------------------------
|
||
|
||
def _kea_subnets(self) -> list[dict[str, Any]]:
|
||
"""Return Kea's managed subnets, or raise if the plugin is absent."""
|
||
try:
|
||
return self._get("/api/kea/dhcpv4/searchSubnet").get("rows") or []
|
||
except Exception as exc:
|
||
raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc
|
||
|
||
def _kea_subnet_uuid_for(self, subnet: str, ip: str) -> str:
|
||
"""Resolve a reservation's target subnet to Kea's subnet UUID.
|
||
|
||
Prefers an exact CIDR match on ``subnet``; falls back to whichever
|
||
managed subnet contains ``ip`` when the caller left ``subnet`` empty.
|
||
"""
|
||
rows = self._kea_subnets()
|
||
|
||
if subnet:
|
||
for row in rows:
|
||
if str(row.get("subnet") or "").strip() == subnet:
|
||
return str(row["uuid"])
|
||
raise ValueError(f"No Kea-managed subnet matches {subnet}")
|
||
|
||
ip_obj = ip_address(ip)
|
||
for row in rows:
|
||
try:
|
||
if ip_obj in ip_network(row["subnet"], strict=False):
|
||
return str(row["uuid"])
|
||
except (ValueError, KeyError):
|
||
continue
|
||
raise ValueError(f"No Kea-managed subnet contains {ip}")
|
||
|
||
def get_dhcp_reservations(self) -> list[dict[str, Any]]:
|
||
"""Return all Kea DHCPv4 static reservations, vendor-neutral.
|
||
|
||
This is the *configured* state — ``get_dhcp_leases()`` returns what
|
||
is actually leased. Kea's ``subnet`` field on a reservation is a
|
||
model relation: depending on version it comes back as the related
|
||
subnet's CIDR or as its UUID, so both are accepted and normalised to
|
||
a CIDR here. An unresolvable relation degrades to an empty string
|
||
rather than raising — one orphaned reservation must not make the
|
||
whole inventory unreadable.
|
||
|
||
:raises RuntimeError: if the Kea plugin isn't installed/enabled.
|
||
"""
|
||
try:
|
||
data = self._post(
|
||
"/api/kea/dhcpv4/searchReservation",
|
||
{"current": 1, "rowCount": -1},
|
||
)
|
||
except Exception as exc:
|
||
raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc
|
||
|
||
rows = data.get("rows") or []
|
||
if not rows:
|
||
return []
|
||
|
||
cidr_by_uuid = {
|
||
str(row.get("uuid") or ""): str(row.get("subnet") or "")
|
||
for row in self._kea_subnets()
|
||
}
|
||
known_cidrs = set(cidr_by_uuid.values())
|
||
|
||
reservations: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
raw_subnet = str(row.get("subnet") or "").strip()
|
||
if raw_subnet in known_cidrs:
|
||
subnet = raw_subnet
|
||
else:
|
||
subnet = cidr_by_uuid.get(raw_subnet, "")
|
||
|
||
reservations.append({
|
||
"uuid": str(row.get("uuid") or ""),
|
||
"mac": str(row.get("hw_address") or ""),
|
||
"ip": str(row.get("ip_address") or ""),
|
||
"hostname": str(row.get("hostname") or ""),
|
||
"description": str(row.get("description") or ""),
|
||
"subnet": subnet,
|
||
})
|
||
return reservations
|
||
|
||
def apply_dhcp_reservation(
|
||
self, reservation: dict[str, Any], *, uuid: str | None = None
|
||
) -> dict[str, Any]:
|
||
"""Create or update a single Kea DHCPv4 static reservation.
|
||
|
||
Does not reconfigure the service — call ``commit_dhcp_reservations()``
|
||
once after a batch, so a ruleset apply costs one daemon reload rather
|
||
than one per reservation.
|
||
|
||
:raises ValueError: if no Kea-managed subnet matches the reservation.
|
||
:raises RuntimeError: if Kea rejects the write.
|
||
"""
|
||
subnet_uuid = self._kea_subnet_uuid_for(
|
||
str(reservation.get("subnet") or "").strip(),
|
||
str(reservation.get("ip") or ""),
|
||
)
|
||
|
||
payload = {
|
||
"reservation": {
|
||
"subnet": subnet_uuid,
|
||
"ip_address": reservation.get("ip", ""),
|
||
"hw_address": reservation.get("mac", ""),
|
||
"hostname": reservation.get("hostname", ""),
|
||
"description": reservation.get("description", ""),
|
||
}
|
||
}
|
||
path = (
|
||
f"/api/kea/dhcpv4/setReservation/{uuid}"
|
||
if uuid
|
||
else "/api/kea/dhcpv4/addReservation"
|
||
)
|
||
result = self._post(path, payload)
|
||
if result.get("result") != "saved":
|
||
raise RuntimeError(
|
||
f"Kea rejected DHCP reservation for {reservation.get('ip')}: {result}"
|
||
)
|
||
return {"success": True}
|
||
|
||
def commit_dhcp_reservations(self) -> dict[str, Any]:
|
||
"""Reload Kea so pending reservation writes take effect."""
|
||
self._post("/api/kea/service/reconfigure")
|
||
return {"success": True}
|
||
|
||
# ── Kea DHCPv4 subnets ────────────────────────────────────────────────
|
||
|
||
# OPNsense renders repeatable option fields ("AsList") as comma-separated
|
||
# strings, and the pool list as a newline-separated text block. Which of
|
||
# the two shapes a given field uses is a property of the model, not of the
|
||
# request, so the mapping is a constant rather than something to sniff.
|
||
_KEA_LIST_OPTIONS = (
|
||
"routers",
|
||
"domain_name_servers",
|
||
"domain_search",
|
||
"ntp_servers",
|
||
)
|
||
_KEA_SCALAR_OPTIONS = ("domain_name",)
|
||
|
||
@staticmethod
|
||
def _kea_read_list(raw: Any) -> list[str]:
|
||
"""Read an OPNsense list field, in either shape it comes back in.
|
||
|
||
Plain string: ``"a,b"``. Selection map: ``{"a": {"selected": 1}, ...}``
|
||
-- which OPNsense uses for some model field types and versions. Both
|
||
appear in the wild for the same logical field, so both are accepted
|
||
rather than pinning the driver to one OPNsense release.
|
||
"""
|
||
if isinstance(raw, dict):
|
||
return [
|
||
str(key)
|
||
for key, meta in raw.items()
|
||
if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1"
|
||
]
|
||
if isinstance(raw, list):
|
||
return [str(item).strip() for item in raw if str(item).strip()]
|
||
return [part.strip() for part in str(raw or "").split(",") if part.strip()]
|
||
|
||
@staticmethod
|
||
def _kea_read_scalar(raw: Any) -> str:
|
||
if isinstance(raw, dict):
|
||
selected = [
|
||
str(key)
|
||
for key, meta in raw.items()
|
||
if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1"
|
||
]
|
||
return selected[0] if selected else ""
|
||
return str(raw or "").strip()
|
||
|
||
def get_dhcp_subnets(self) -> list[dict[str, Any]]:
|
||
"""Return every Kea DHCPv4 subnet with its pools and options.
|
||
|
||
``searchSubnet`` only carries uuid/subnet/description, so each row is
|
||
followed by a ``getSubnet`` call for the option data. That is one
|
||
request per subnet; a firewall serves a handful of them, so the extra
|
||
round trips cost less than the reconfigure they help avoid.
|
||
|
||
An option Kea does not carry is *omitted* from ``option_data`` rather
|
||
than reported as empty. The generic diff treats an absent key as "not
|
||
managed", so reporting ``[]`` here would make every unmanaged option
|
||
look like a pending change and reload the daemon on every run.
|
||
|
||
A subnet whose detail fetch fails is skipped with a log line instead
|
||
of aborting: one broken record must not make the whole inventory
|
||
unreadable, same rule as ``get_dhcp_reservations()``.
|
||
|
||
:raises RuntimeError: if the Kea plugin isn't installed/enabled.
|
||
"""
|
||
rows = self._kea_subnets()
|
||
|
||
subnets: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
uuid = str(row.get("uuid") or "")
|
||
if not uuid:
|
||
continue
|
||
try:
|
||
detail = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}")
|
||
except Exception as exc:
|
||
logger.warning("Kea subnet %s detail fetch failed, skipping: %s", uuid, exc)
|
||
continue
|
||
|
||
record = detail.get("subnet") or {}
|
||
raw_options = record.get("option_data") or {}
|
||
|
||
option_data: dict[str, Any] = {}
|
||
for name in self._KEA_LIST_OPTIONS:
|
||
values = self._kea_read_list(raw_options.get(name))
|
||
if values:
|
||
option_data[name] = values
|
||
for name in self._KEA_SCALAR_OPTIONS:
|
||
value = self._kea_read_scalar(raw_options.get(name))
|
||
if value:
|
||
option_data[name] = value
|
||
|
||
pools = [
|
||
line.strip()
|
||
for line in str(record.get("pools") or "").splitlines()
|
||
if line.strip()
|
||
]
|
||
|
||
subnets.append({
|
||
"uuid": uuid,
|
||
"subnet": str(record.get("subnet") or row.get("subnet") or ""),
|
||
"description": str(record.get("description") or ""),
|
||
"pools": pools,
|
||
"option_data": option_data,
|
||
"match_client_id": str(record.get("match-client-id") or "0") == "1",
|
||
})
|
||
return subnets
|
||
|
||
def apply_dhcp_subnet(
|
||
self, subnet: dict[str, Any], *, uuid: str | None = None
|
||
) -> dict[str, Any]:
|
||
"""Create or update a single Kea DHCPv4 subnet.
|
||
|
||
``option_data`` is a partial update, as the mixin contract requires:
|
||
on an update the subnet's current options are read first and only the
|
||
named ones are replaced. Without that, managing `domain_search` alone
|
||
would blank the `routers` Kea autocollected and strand every client on
|
||
that VLAN without a gateway.
|
||
|
||
Setting any option explicitly also turns ``option_data_autocollect``
|
||
off. Left on, Kea keeps re-filling routers/DNS/NTP from the interface
|
||
and the next diff sees a change again -- a reconfigure loop rather
|
||
than a converged state.
|
||
|
||
Does not reconfigure the service -- call ``commit_dhcp_subnets()``
|
||
once after a batch.
|
||
|
||
:raises RuntimeError: if Kea rejects the write.
|
||
"""
|
||
desired_options = dict(subnet.get("option_data") or {})
|
||
|
||
merged_options: dict[str, str] = {}
|
||
if uuid and desired_options:
|
||
try:
|
||
current = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}")
|
||
raw = (current.get("subnet") or {}).get("option_data") or {}
|
||
except Exception as exc:
|
||
raise RuntimeError(
|
||
f"Kea subnet {uuid} could not be read before update: {exc}"
|
||
) from exc
|
||
for name in self._KEA_LIST_OPTIONS:
|
||
values = self._kea_read_list(raw.get(name))
|
||
if values:
|
||
merged_options[name] = ",".join(values)
|
||
for name in self._KEA_SCALAR_OPTIONS:
|
||
value = self._kea_read_scalar(raw.get(name))
|
||
if value:
|
||
merged_options[name] = value
|
||
|
||
for name, value in desired_options.items():
|
||
merged_options[name] = ",".join(value) if isinstance(value, list) else str(value)
|
||
|
||
record: dict[str, Any] = {"subnet": subnet.get("subnet", "")}
|
||
if subnet.get("description") is not None:
|
||
record["description"] = subnet.get("description") or ""
|
||
if subnet.get("pools") is not None:
|
||
record["pools"] = "\n".join(subnet.get("pools") or [])
|
||
if subnet.get("match_client_id") is not None:
|
||
record["match-client-id"] = "1" if subnet.get("match_client_id") else "0"
|
||
if desired_options:
|
||
record["option_data"] = merged_options
|
||
record["option_data_autocollect"] = "0"
|
||
|
||
path = (
|
||
f"/api/kea/dhcpv4/setSubnet/{uuid}" if uuid else "/api/kea/dhcpv4/addSubnet"
|
||
)
|
||
result = self._post(path, {"subnet": record})
|
||
if result.get("result") != "saved":
|
||
raise RuntimeError(
|
||
f"Kea rejected DHCP subnet {subnet.get('subnet')}: {result}"
|
||
)
|
||
return {"success": True}
|
||
|
||
def commit_dhcp_subnets(self) -> dict[str, Any]:
|
||
"""Reload Kea so pending subnet writes take effect."""
|
||
self._post("/api/kea/service/reconfigure")
|
||
return {"success": True}
|
||
|
||
def get_services(self) -> list[dict[str, Any]]:
|
||
"""Return running services from OPNsense.
|
||
|
||
Calls ``GET /api/core/service/search`` and normalises the rows to
|
||
``{name, running, enabled, pid}`` — the same format used by the
|
||
OpenWrt driver so the UI can render them identically.
|
||
"""
|
||
data = self._get("/api/core/service/search")
|
||
result: list[dict[str, Any]] = []
|
||
for row in data.get("rows", []):
|
||
result.append({
|
||
"name": row.get("name") or row.get("id", ""),
|
||
"running": bool(row.get("running", 0)),
|
||
"enabled": True, # OPNsense has no separate enabled/disabled state
|
||
"pid": 0,
|
||
})
|
||
return sorted(result, key=lambda x: x["name"].lower())
|
||
|
||
def manage_service(self, name: str, action: str) -> dict[str, Any]:
|
||
"""Execute a lifecycle action on an OPNsense service.
|
||
|
||
OPNsense exposes per-plugin service endpoints at
|
||
``/api/<plugin>/service/<action>``. The ``name`` parameter must
|
||
match the service ``id`` returned by :meth:`get_services`.
|
||
|
||
Supported actions: ``start``, ``stop``, ``restart``.
|
||
``enable`` and ``disable`` are not supported on OPNsense (plugins
|
||
are enabled/disabled by installing or removing them).
|
||
"""
|
||
import re as _re
|
||
if not _re.match(r'^[a-zA-Z0-9_\-]+$', name):
|
||
raise ValueError(f"Invalid service name: {name!r}")
|
||
if action not in ('start', 'stop', 'restart'):
|
||
return {"success": False, "output": f"Action '{action}' is not supported on OPNsense services"}
|
||
try:
|
||
result = self._post(f"/api/{name}/service/{action}")
|
||
return {"success": True, "output": str(result)}
|
||
except Exception as exc:
|
||
logger.warning("Service action failed: %s", exc)
|
||
return {"success": False, "output": str(exc)}
|
||
|
||
def get_vpn_tunnels(self) -> dict[str, dict[str, Any]]:
|
||
"""Return status of all configured VPN tunnels.
|
||
|
||
Queries IPsec, OpenVPN, and WireGuard in order and merges results
|
||
into a single dict keyed by a unique tunnel identifier.
|
||
|
||
Each entry follows the ``VPNTunnelDict`` schema:
|
||
``type``, ``local_endpoint``, ``remote_endpoint``, ``is_up``,
|
||
``uptime``, ``bytes_in``, ``bytes_out``, ``description``.
|
||
|
||
* **IPsec** — ``GET /api/ipsec/sessions``
|
||
Falls back to ``GET /api/ipsec/leases/searchPhase2`` on older
|
||
OPNsense releases.
|
||
* **OpenVPN** — ``GET /api/openvpn/instances/search``
|
||
Instance status obtained from ``GET /api/openvpn/service/show``.
|
||
* **WireGuard** — ``GET /api/wireguard/service/show``
|
||
"""
|
||
tunnels: dict[str, dict[str, Any]] = {}
|
||
|
||
# ── IPsec ──────────────────────────────────────────────────────────
|
||
try:
|
||
data = self._get("/api/ipsec/sessions")
|
||
# OPNsense 24.x returns {"response": [...]} or a bare list
|
||
sessions = data.get("response") if isinstance(data, dict) else data
|
||
if isinstance(sessions, list):
|
||
for session in sessions:
|
||
# Each session object may contain multiple child SAs; we
|
||
# report one entry per IKE peer.
|
||
remote_host = (
|
||
session.get("remote-host")
|
||
or session.get("remote_host")
|
||
or session.get("remote-id")
|
||
or ""
|
||
)
|
||
local_host = (
|
||
session.get("local-host")
|
||
or session.get("local_host")
|
||
or session.get("local-id")
|
||
or ""
|
||
)
|
||
name = (
|
||
session.get("uniqueid")
|
||
or session.get("con-id")
|
||
or session.get("name")
|
||
or remote_host
|
||
or f"ipsec-{len(tunnels)}"
|
||
)
|
||
state = str(session.get("state") or session.get("ikey-state") or "").lower()
|
||
is_up = state in ("established", "up", "installed")
|
||
uptime_raw = session.get("established") or session.get("uptime") or 0
|
||
try:
|
||
uptime = int(uptime_raw)
|
||
except (ValueError, TypeError):
|
||
uptime = 0
|
||
# Byte counters may live in child-sa entries
|
||
bytes_in = 0
|
||
bytes_out = 0
|
||
for child in session.get("child-sas", {}).values() if isinstance(session.get("child-sas"), dict) else []:
|
||
try:
|
||
bytes_in += int(child.get("bytes-in", 0) or 0)
|
||
bytes_out += int(child.get("bytes-out", 0) or 0)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
description = (
|
||
session.get("local-id")
|
||
or session.get("description")
|
||
or ""
|
||
)
|
||
key = f"ipsec-{name}"
|
||
tunnels[key] = {
|
||
"type": "IPsec",
|
||
"local_endpoint": local_host,
|
||
"remote_endpoint": remote_host,
|
||
"is_up": is_up,
|
||
"uptime": uptime,
|
||
"bytes_in": bytes_in,
|
||
"bytes_out": bytes_out,
|
||
"description": description,
|
||
}
|
||
except Exception as exc:
|
||
logger.debug("IPsec session status fetch failed: %s", exc)
|
||
# Try legacy Phase-2 leases endpoint (OPNsense < 23.x)
|
||
try:
|
||
data = self._post(
|
||
"/api/ipsec/leases/searchPhase2",
|
||
{"current": 1, "rowCount": -1, "searchPhrase": "", "sort": {}},
|
||
)
|
||
for row in data.get("rows", []):
|
||
name = row.get("id") or row.get("con") or f"ipsec-{len(tunnels)}"
|
||
key = f"ipsec-{name}"
|
||
state = str(row.get("state") or "").lower()
|
||
is_up = state in ("established", "installed", "up")
|
||
tunnels[key] = {
|
||
"type": "IPsec",
|
||
"local_endpoint": row.get("local-ts", ""),
|
||
"remote_endpoint": row.get("remote-ts", ""),
|
||
"is_up": is_up,
|
||
"uptime": 0,
|
||
"bytes_in": int(row.get("bytes-in", 0) or 0),
|
||
"bytes_out": int(row.get("bytes-out", 0) or 0),
|
||
"description": row.get("con", ""),
|
||
}
|
||
except Exception as exc:
|
||
logger.debug("Legacy IPsec Phase-2 lease fetch failed: %s", exc)
|
||
|
||
# ── OpenVPN ────────────────────────────────────────────────────────
|
||
try:
|
||
data = self._get("/api/openvpn/instances/search")
|
||
instances = data.get("rows", [])
|
||
# Fetch live service status to determine up/down state
|
||
try:
|
||
show = self._get("/api/openvpn/service/show")
|
||
# show is a dict of {instance_id: {status, ...}}
|
||
status_map: dict[str, Any] = show if isinstance(show, dict) else {}
|
||
except Exception as exc:
|
||
logger.debug("OpenVPN service status fetch failed: %s", exc)
|
||
status_map = {}
|
||
for inst in instances:
|
||
iid = inst.get("id") or inst.get("vpnid") or f"ovpn-{len(tunnels)}"
|
||
name = inst.get("description") or inst.get("dev") or str(iid)
|
||
key = f"openvpn-{iid}"
|
||
svc = status_map.get(str(iid), {})
|
||
is_up = str(svc.get("running", False)).lower() in ("true", "1", "yes")
|
||
local_addr = inst.get("local") or inst.get("interface") or ""
|
||
remote_addr = inst.get("server") or inst.get("remote") or ""
|
||
tunnels[key] = {
|
||
"type": "SSL",
|
||
"local_endpoint": local_addr,
|
||
"remote_endpoint": remote_addr,
|
||
"is_up": is_up,
|
||
"uptime": 0,
|
||
"bytes_in": 0,
|
||
"bytes_out": 0,
|
||
"description": name,
|
||
}
|
||
except Exception as exc:
|
||
logger.debug("OpenVPN instance fetch failed (plugin not installed?): %s", exc)
|
||
|
||
# ── WireGuard ──────────────────────────────────────────────────────
|
||
try:
|
||
# /api/wireguard/service/show returns structured JSON:
|
||
# {"total": N, "rows": [
|
||
# {"if":"wg0", "type":"interface", ...},
|
||
# {"if":"wg0", "type":"peer", "public-key":"...",
|
||
# "endpoint":"1.2.3.4:51820", "transfer-rx":N, "transfer-tx":N,
|
||
# "name":"HCQ", "peer-status":"online",
|
||
# "latest-handshake-age":45, "ifname":"HCQ"}, ...
|
||
# ]}
|
||
data = self._get("/api/wireguard/service/show")
|
||
rows = data.get("rows", []) if isinstance(data, dict) else []
|
||
|
||
wg_idx = 0
|
||
for row in rows:
|
||
if row.get("type") != "peer":
|
||
continue
|
||
wg_idx += 1
|
||
|
||
endpoint = (row.get("endpoint") or "").strip()
|
||
if endpoint and endpoint != "(none)":
|
||
remote_ip = endpoint.rsplit(":", 1)[0].strip("[]")
|
||
else:
|
||
remote_ip = ""
|
||
|
||
is_up = row.get("peer-status") == "online"
|
||
bytes_in = int(row.get("transfer-rx") or 0)
|
||
bytes_out = int(row.get("transfer-tx") or 0)
|
||
hs_age = row.get("latest-handshake-age")
|
||
uptime = int(hs_age) if hs_age else 0
|
||
|
||
description = (
|
||
(row.get("name") or "").strip()
|
||
or (row.get("ifname") or "").strip()
|
||
or f"wg-peer-{wg_idx}"
|
||
)
|
||
|
||
tunnels[f"wireguard-{wg_idx}"] = {
|
||
"type": "WireGuard",
|
||
"local_endpoint": "",
|
||
"remote_endpoint": remote_ip,
|
||
"is_up": is_up,
|
||
"uptime": uptime,
|
||
"bytes_in": bytes_in,
|
||
"bytes_out": bytes_out,
|
||
"description": description,
|
||
}
|
||
except Exception as exc:
|
||
logger.debug("WireGuard status fetch failed (plugin not installed?): %s", exc)
|
||
|
||
return tunnels
|
||
|
||
def get_available_updates(self) -> list[dict[str, Any]]:
|
||
"""Return available firmware and package updates.
|
||
|
||
Triggers an async update-check on OPNsense via
|
||
``POST /api/core/firmware/check``, then polls
|
||
``GET /api/core/firmware/status`` for up to 15 seconds.
|
||
Returns a list of ``{name, current_version, new_version}`` dicts,
|
||
or an empty list when everything is up to date or the check has
|
||
not yet finished.
|
||
"""
|
||
import time
|
||
try:
|
||
self._post("/api/core/firmware/check")
|
||
except Exception as exc:
|
||
logger.debug("Firmware update check trigger failed: %s", exc)
|
||
|
||
for _ in range(5):
|
||
time.sleep(3)
|
||
try:
|
||
status = self._get("/api/core/firmware/status")
|
||
state = status.get("status", "none")
|
||
if state in ("update", "upgrade"):
|
||
updates = (
|
||
status.get("upgrade_packages")
|
||
or status.get("updates")
|
||
or []
|
||
)
|
||
return [
|
||
{
|
||
"name": u.get("name", ""),
|
||
"current_version": u.get("current_version", u.get("version", "")),
|
||
"new_version": u.get("new_version", u.get("version", "")),
|
||
}
|
||
for u in updates
|
||
]
|
||
if state == "latest":
|
||
return []
|
||
except Exception as exc:
|
||
logger.debug("Firmware status poll failed: %s", exc)
|
||
return []
|
||
|
||
def get_device_warnings(self) -> list[dict[str, Any]]:
|
||
"""Return a list of warning dicts for issues detected on this device.
|
||
|
||
Reads the cached ``GET /api/core/firmware/status`` (no network
|
||
update trigger) to detect available package/firmware updates.
|
||
"""
|
||
warnings: list[dict[str, Any]] = []
|
||
try:
|
||
status = self._get("/api/core/firmware/status")
|
||
state = status.get("status", "none")
|
||
if state in ("update", "upgrade"):
|
||
upgrades = (
|
||
status.get("upgrade_packages")
|
||
or status.get("updates")
|
||
or []
|
||
)
|
||
if upgrades:
|
||
warnings.append({
|
||
"code": "updates_available",
|
||
"meta": {
|
||
"count": len(upgrades),
|
||
"packages": [u.get("name", "") for u in upgrades[:10]],
|
||
},
|
||
})
|
||
except Exception as exc:
|
||
logger.debug("Failed to check firmware status for device warnings: %s", exc)
|
||
return warnings
|
||
|
||
def apply_updates(self, packages: list[str]) -> dict[str, Any]:
|
||
"""Trigger a full firmware upgrade on OPNsense.
|
||
|
||
Note: OPNsense upgrades the entire system at once rather than
|
||
individual packages. The ``packages`` parameter is accepted for
|
||
API compatibility but is ignored — the full upgrade is always
|
||
applied.
|
||
|
||
Calls ``POST /api/core/firmware/upgrade``.
|
||
"""
|
||
try:
|
||
result = self._post("/api/core/firmware/upgrade")
|
||
return {"success": True, "output": str(result)}
|
||
except Exception as exc:
|
||
logger.warning("Firmware upgrade failed: %s", exc)
|
||
return {"success": False, "output": str(exc)}
|
||
|
||
def search_packages(self, query: str) -> list[dict[str, Any]]:
|
||
"""Search available OPNsense plugins by name or description.
|
||
|
||
Filters the full plugin list from ``GET /api/core/firmware/info``
|
||
against *query* (case-insensitive substring match on name and
|
||
comment). Returns all matching plugins with an ``installed``
|
||
flag so the UI can show which ones are already active.
|
||
"""
|
||
q = query.lower()
|
||
info = self._get("/api/core/firmware/info")
|
||
result: list[dict[str, Any]] = []
|
||
for p in info.get("plugin", []):
|
||
name = p.get("name", "")
|
||
comment = p.get("comment", "")
|
||
if q in name.lower() or q in comment.lower():
|
||
result.append({
|
||
"name": name,
|
||
"version": p.get("version", ""),
|
||
"installed": p.get("installed") == "1",
|
||
"description": comment,
|
||
"size": 0,
|
||
"source": "opnsense-plugins",
|
||
})
|
||
return sorted(result, key=lambda x: x["name"].lower())
|
||
|
||
def install_package(self, name: str) -> dict[str, Any]:
|
||
"""Install an OPNsense plugin by name.
|
||
|
||
Calls ``POST /api/core/firmware/install/{name}``.
|
||
"""
|
||
import re as _re
|
||
if not _re.match(r'^[a-zA-Z0-9_\-\.]+$', name):
|
||
raise ValueError(f"Invalid package name: {name!r}")
|
||
try:
|
||
result = self._post(f"/api/core/firmware/install/{name}")
|
||
return {"success": True, "output": str(result)}
|
||
except Exception as exc:
|
||
logger.warning("Package install failed: %s", exc)
|
||
return {"success": False, "output": str(exc)}
|
||
|
||
def uninstall_package(self, name: str) -> dict[str, Any]:
|
||
"""Remove an OPNsense plugin by name.
|
||
|
||
Calls ``POST /api/core/firmware/remove/{name}``.
|
||
"""
|
||
import re as _re
|
||
if not _re.match(r'^[a-zA-Z0-9_\-\.]+$', name):
|
||
raise ValueError(f"Invalid package name: {name!r}")
|
||
try:
|
||
result = self._post(f"/api/core/firmware/remove/{name}")
|
||
return {"success": True, "output": str(result)}
|
||
except Exception as exc:
|
||
logger.warning("Package uninstall failed: %s", exc)
|
||
return {"success": False, "output": str(exc)}
|
||
|
||
# ── SNMP / Health ──────────────────────────────────────────────────────────
|
||
|
||
def get_snmp_config(self):
|
||
"""Return SNMP config if the os-net-snmp plugin is installed and enabled.
|
||
|
||
Uses GET /api/netsnmp/general/get (os-net-snmp plugin API).
|
||
Returns None if the plugin is not installed or SNMP is disabled.
|
||
"""
|
||
try:
|
||
from napalm_device_types.models import SNMPConfigDict
|
||
except ImportError:
|
||
return None
|
||
|
||
try:
|
||
# Short timeout — endpoint may not exist if os-net-snmp plugin is not installed
|
||
url = self.base_url.rstrip("/") + "/api/netsnmp/general/get"
|
||
response = self.session.get(url, timeout=5)
|
||
if response.status_code != 200:
|
||
return None
|
||
data = response.json()
|
||
except Exception as exc:
|
||
logger.debug("SNMP config fetch failed (plugin not installed?): %s", exc)
|
||
return None
|
||
|
||
if not data:
|
||
return None
|
||
|
||
general = data.get("general", data)
|
||
enabled = str(general.get("enabled", "0")) == "1"
|
||
if not enabled:
|
||
return None
|
||
|
||
community = general.get("community", "public") or "public"
|
||
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
||
|
||
# ── Wake-on-LAN ──────────────────────────────────────────────────────────────
|
||
|
||
def send_wake_on_lan(self, mac_address: str, interface: str = "") -> dict[str, Any]:
|
||
"""Send a Wake-on-LAN magic packet via OPNsense's os-wol plugin.
|
||
|
||
Calls ``POST /api/wol/wol/set`` with an ephemeral (non-persisted) entry —
|
||
omitting ``uuid`` from the payload makes OPNsense's ``WolController::setAction``
|
||
validate and wake immediately without saving a host to config.xml. Requires
|
||
the os-wol plugin to be installed and the target interface to have a static
|
||
IPv4 address configured (OPNsense computes the broadcast address from the
|
||
interface's own IP/subnet; DHCP-assigned interfaces are rejected).
|
||
|
||
:param interface: OPNsense's *assigned* interface identifier (e.g. "lan",
|
||
"opt1") — NOT the physical device name returned by get_interfaces()/
|
||
get_networks() (e.g. "em0"). Required by this driver.
|
||
:raises ValueError: if interface is not provided.
|
||
"""
|
||
if not interface:
|
||
raise ValueError("OPNsense requires an interface for Wake-on-LAN")
|
||
try:
|
||
result = self._post(
|
||
"/api/wol/wol/set",
|
||
{"wake": {"interface": interface, "mac": mac_address.strip()}},
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Wake-on-LAN send failed for %s via %s: %s", mac_address, interface, exc)
|
||
return {"success": False, "output": str(exc)}
|
||
|
||
status = result.get("status")
|
||
if status == "error":
|
||
return {
|
||
"success": False,
|
||
"output": result.get("error_msg", "OPNsense rejected the Wake-on-LAN request"),
|
||
}
|
||
if not result:
|
||
# Model validation (malformed MAC/interface) fails silently with an
|
||
# empty {} response at HTTP 200 — no error_msg to surface.
|
||
return {
|
||
"success": False,
|
||
"output": "OPNsense rejected the request (check interface identifier and MAC format)",
|
||
}
|
||
return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"}
|
||
|
||
def run_device_action(self, action: str) -> dict[str, Any]:
|
||
"""Execute a named action on the firewall."""
|
||
if action == "fix_snmp":
|
||
return self._action_fix_snmp()
|
||
raise NotImplementedError(f"Unknown action: {action!r}")
|
||
|
||
def _action_fix_snmp(self) -> dict[str, Any]:
|
||
"""Install os-net-snmp plugin, configure community 'public', open firewall.
|
||
|
||
Steps:
|
||
1. Install os-net-snmp via firmware API (idempotent)
|
||
2. Configure via POST /api/netsnmp/general/set
|
||
3. Start/restart the service
|
||
4. Add a floating firewall rule allowing UDP/161 from any source
|
||
(OPNsense default-drops traffic arriving on non-LAN interfaces
|
||
such as WireGuard; without this rule SNMP is unreachable from
|
||
management networks even though the daemon is running)
|
||
5. Apply firewall and verify
|
||
"""
|
||
lines: list = []
|
||
|
||
# 1. Install os-net-snmp plugin ONLY if not already present.
|
||
# Calling firmware/install on an already-installed plugin triggers an
|
||
# async reinstall that overwrites the config with factory defaults a few
|
||
# minutes later — causing SNMP to stop working again after the fix.
|
||
plugin_installed = False
|
||
try:
|
||
chk = self._get("/api/netsnmp/general/get")
|
||
plugin_installed = isinstance(chk, dict) and "general" in chk
|
||
except Exception:
|
||
pass
|
||
|
||
if not plugin_installed:
|
||
try:
|
||
result = self._post("/api/core/firmware/install/os-net-snmp")
|
||
lines.append(f"[install] {result}")
|
||
import time as _time
|
||
_time.sleep(5) # wait for install to settle
|
||
except Exception as exc:
|
||
logger.debug("os-net-snmp install failed: %s", exc)
|
||
lines.append(f"[install] failed: {exc}")
|
||
else:
|
||
lines.append("[install] os-net-snmp already installed — skipping reinstall")
|
||
|
||
# 2. Configure SNMP
|
||
# The listen field must contain the management IP so that snmpd binds
|
||
# to the correct interface. Without it snmpd may not respond on any
|
||
# non-loopback address.
|
||
# The API expects {"<ip>": {"selected": 1}} — we use self.hostname
|
||
# which is always the device's management IP in netOrk.
|
||
listen_val: dict = {self.hostname: {"selected": 1}}
|
||
try:
|
||
self._post("/api/netsnmp/general/set", {
|
||
"general": {
|
||
"enabled": "1",
|
||
"community": "public",
|
||
"contact": "netork@localhost",
|
||
"location": "Managed by netOrk",
|
||
"sysobjid": "",
|
||
"bindip": "",
|
||
"listen": listen_val,
|
||
}
|
||
})
|
||
lines.append(f"[config] SNMP enabled with community 'public', listen={self.hostname}.")
|
||
except Exception as exc:
|
||
logger.debug("SNMP config failed: %s", exc)
|
||
lines.append(f"[config] error: {exc}")
|
||
|
||
# 3. Start / restart the SNMP service
|
||
try:
|
||
self._post("/api/netsnmp/service/restart")
|
||
lines.append("[service] net-snmp restarted.")
|
||
except Exception as exc:
|
||
logger.debug("SNMP service restart failed, trying start: %s", exc)
|
||
try:
|
||
self._post("/api/netsnmp/service/start")
|
||
lines.append("[service] net-snmp started.")
|
||
except Exception as exc2:
|
||
lines.append(f"[service] start failed: {exc2}")
|
||
|
||
# 4. Add floating firewall rule for SNMP (UDP/161 → self)
|
||
# OPNsense blocks traffic arriving on non-LAN interfaces by default.
|
||
# A floating pass rule makes SNMP reachable from all management nets.
|
||
try:
|
||
existing = self._post("/api/firewall/filter/searchRule",
|
||
{"current": 1, "rowCount": -1, "searchPhrase": "[netork] Allow SNMP"})
|
||
if not (existing.get("rows") or []):
|
||
self._post("/api/firewall/filter/addRule", {
|
||
"rule": {
|
||
"enabled": "1",
|
||
"sequence": "1",
|
||
"action": "pass",
|
||
"quick": "1",
|
||
"interface": "", # floating — all interfaces
|
||
"direction": "in",
|
||
"ipprotocol": "inet",
|
||
"protocol": "udp",
|
||
"source_net": "any",
|
||
"source_not": "0",
|
||
"destination_net": "(self)",
|
||
"destination_not": "0",
|
||
"destination_port": "161",
|
||
"log": "0",
|
||
"floating": "yes",
|
||
"descr": "[netork] Allow SNMP",
|
||
}
|
||
})
|
||
lines.append("[firewall] Floating pass rule added for UDP/161.")
|
||
else:
|
||
lines.append("[firewall] Pass rule already present.")
|
||
self._post("/api/firewall/filter/apply", {})
|
||
lines.append("[firewall] Rules applied.")
|
||
except Exception as exc:
|
||
logger.debug("Firewall rule for SNMP failed: %s", exc)
|
||
lines.append(f"[firewall] error: {exc}")
|
||
|
||
# 5. Verify — try actual UDP/161 probe first, fall back to API config check
|
||
success = False
|
||
try:
|
||
import socket as _socket
|
||
sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
|
||
sock.settimeout(3)
|
||
community = b"public"
|
||
# Minimal SNMPv2c GetRequest for sysDescr
|
||
pdu = (b"\x30\x26\x02\x01\x01\x04"
|
||
+ bytes([len(community)]) + community
|
||
+ b"\xa0\x19\x02\x04\x00\x00\x00\x01"
|
||
+ b"\x02\x01\x00\x02\x01\x00"
|
||
+ b"\x30\x0b\x30\x09\x06\x05\x2b\x06\x01\x02\x01\x05\x00")
|
||
sock.sendto(pdu, (self.hostname, 161))
|
||
try:
|
||
data, _ = sock.recvfrom(1024)
|
||
success = len(data) > 0
|
||
lines.append("[ok] SNMP UDP probe successful.")
|
||
except _socket.timeout:
|
||
lines.append("[warn] SNMP UDP probe timed out — service may be starting.")
|
||
finally:
|
||
sock.close()
|
||
except Exception as exc:
|
||
logger.debug("SNMP UDP probe failed: %s", exc)
|
||
# Fall back to API config check
|
||
try:
|
||
cfg = self._get("/api/netsnmp/general/get")
|
||
general = cfg.get("general", cfg)
|
||
success = str(general.get("enabled", "0")) == "1" and bool(general.get("community"))
|
||
lines.append("[ok] SNMP config active (UDP probe unavailable)." if success
|
||
else "[warn] SNMP config check failed.")
|
||
except Exception:
|
||
lines.append("[warn] Could not verify SNMP state.")
|
||
|
||
return {"success": success, "output": "\n".join(lines)}
|
||
|
||
def get_firewall_aliases(self) -> list[dict[str, Any]]:
|
||
"""Return all firewall aliases, sorted by type then name.
|
||
|
||
Each entry contains:
|
||
* ``name`` — alias name
|
||
* ``type`` — alias type (host, network, port, url, urltable, geoip, etc.)
|
||
* ``description`` — human-readable description
|
||
* ``content`` — list of values (IPs, networks, ports, URLs, …)
|
||
* ``enabled`` — bool
|
||
* ``counters`` — optional dict with packet/byte stats if available
|
||
"""
|
||
try:
|
||
resp = self._get("/api/firewall/alias/searchItem?current=1&rowCount=-1")
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch firewall aliases: %s", exc)
|
||
return []
|
||
|
||
rows = resp.get("rows") or []
|
||
result = []
|
||
for row in rows:
|
||
content_raw = row.get("content", "") or ""
|
||
# OPNsense stores content as newline-separated values
|
||
content = [v.strip() for v in content_raw.splitlines() if v.strip()]
|
||
result.append({
|
||
"name": row.get("name", ""),
|
||
"type": row.get("type", ""),
|
||
"description": row.get("description", "") or "",
|
||
"content": content,
|
||
"enabled": str(row.get("enabled", "1")) == "1",
|
||
"proto": row.get("proto", "") or "",
|
||
})
|
||
|
||
return sorted(result, key=lambda x: (x["type"], x["name"].lower()))
|
||
|
||
def get_firewall_rules(self) -> list[dict[str, Any]]:
|
||
"""Return all firewall filter rules with interface labels.
|
||
|
||
Extra fields beyond NAPALM standard:
|
||
* ``floating`` — bool, rule applies across all interfaces
|
||
* ``interface_label`` — human-readable interface description
|
||
* ``is_group`` — bool, interface is an interface group
|
||
"""
|
||
try:
|
||
resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1")
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch firewall rules: %s", exc)
|
||
return []
|
||
|
||
# OPNsense provides all human-readable values via %-prefixed fields —
|
||
# no separate lookup needed for interface labels or category names.
|
||
|
||
# Interface group names (to distinguish groups from plain interfaces)
|
||
group_names: set = set()
|
||
try:
|
||
grp = self._get("/api/ifgroups/ifgroups/searchItem?current=1&rowCount=-1")
|
||
for g in (grp.get("rows") or []):
|
||
n = g.get("ifname") or g.get("name") or ""
|
||
if n:
|
||
group_names.add(n)
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch interface groups: %s", exc)
|
||
|
||
rows = resp.get("rows") or []
|
||
result = []
|
||
for row in rows:
|
||
iface = row.get("interface", "") or ""
|
||
iface_list = [i.strip() for i in iface.split(",") if i.strip()]
|
||
# %interface already has the resolved friendly names (e.g. "MGMT, wgadmin")
|
||
iface_label = (row.get("%interface") or "").strip() or ", ".join(iface_list)
|
||
|
||
explicit_floating = (
|
||
str(row.get("floating", "0")) in ("1", "yes")
|
||
or row.get("floating") is True
|
||
)
|
||
floating = explicit_floating or len(iface_list) > 1
|
||
|
||
# %categories is the resolved category name; categories is the UUID
|
||
category = (row.get("%categories") or row.get("category") or "").strip()
|
||
|
||
# Use %-prefixed display values for source/destination
|
||
source_net = (row.get("%source_net") or row.get("source_net") or "any").strip()
|
||
destination_net = (row.get("%destination_net") or row.get("destination_net") or "any").strip()
|
||
|
||
result.append({
|
||
"uuid": row.get("uuid", ""),
|
||
"sequence": int(row.get("sequence", 0) or 0),
|
||
"action": row.get("action", "pass"),
|
||
"quick": str(row.get("quick", "1")) == "1",
|
||
"interface": iface,
|
||
"interface_label": iface_label,
|
||
"floating": floating,
|
||
"is_group": len(iface_list) == 1 and iface_list[0] in group_names,
|
||
"direction": row.get("direction", "in"),
|
||
"ipprotocol": row.get("ipprotocol", "inet"),
|
||
"protocol": row.get("protocol", "any") or "any",
|
||
"source_net": source_net,
|
||
"source_port": row.get("source_port", "") or "",
|
||
"source_not": str(row.get("source_not", "0")) == "1",
|
||
"destination_net": destination_net,
|
||
"destination_port": row.get("destination_port", "") or "",
|
||
"destination_not": str(row.get("destination_not", "0")) == "1",
|
||
"description": row.get("description", "") or "",
|
||
"log": str(row.get("log", "0")) == "1",
|
||
"enabled": str(row.get("enabled", "1")) == "1",
|
||
"category": category,
|
||
})
|
||
|
||
return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))
|
||
|
||
def apply_firewall_rule(self, rule: dict[str, Any], *, uuid: str | None = None) -> dict[str, Any]:
|
||
"""Create or update a single OPNsense firewall filter rule.
|
||
|
||
`rule` uses the vendor-neutral field names from
|
||
``napalm_device_types.models.FirewallRuleDict`` (see
|
||
``FirewallDriver.diff_firewall_rules``/``apply_firewall_ruleset``,
|
||
the generic reconciliation engine that calls this method). This is
|
||
the OPNsense-specific half: translating those fields into the
|
||
``/api/firewall/filter/addRule``/``setRule`` payload shape (string
|
||
"1"/"0" booleans, empty ``interface`` means a floating rule — same
|
||
payload shape as the SNMP self-provisioning rule in
|
||
``_action_fix_snmp``).
|
||
"""
|
||
payload = {
|
||
"rule": {
|
||
"enabled": "1" if rule.get("enabled", True) else "0",
|
||
"sequence": "1",
|
||
"action": rule.get("action", "pass"),
|
||
"quick": "1" if rule.get("quick", True) else "0",
|
||
"interface": rule.get("interface", "") or "",
|
||
"direction": rule.get("direction", "in"),
|
||
"ipprotocol": "inet",
|
||
"protocol": rule.get("protocol", "any"),
|
||
"source_net": rule.get("source_net", "any") or "any",
|
||
"source_port": rule.get("source_port", "") or "",
|
||
"destination_net": rule.get("destination_net", "any") or "any",
|
||
"destination_port": rule.get("destination_port", "") or "",
|
||
"log": "1" if rule.get("log", False) else "0",
|
||
"floating": "yes" if not rule.get("interface") else "no",
|
||
"descr": rule.get("description", ""),
|
||
}
|
||
}
|
||
path = f"/api/firewall/filter/setRule/{uuid}" if uuid else "/api/firewall/filter/addRule"
|
||
return self._post(path, payload)
|
||
|
||
def commit_firewall_rules(self) -> dict[str, Any]:
|
||
"""Reload the firewall filter to activate pending rule changes.
|
||
|
||
Final step after one or more `apply_firewall_rule()` calls — same
|
||
as the last step of `_action_fix_snmp`.
|
||
"""
|
||
return self._post("/api/firewall/filter/apply", {})
|
||
|
||
# ------------------------------------------------------------------
|
||
# Hostname management
|
||
# ------------------------------------------------------------------
|
||
|
||
def set_hostname(self, new_hostname: str) -> None:
|
||
"""Set the system hostname on OPNsense and regenerate the web GUI certificate.
|
||
|
||
Accepts either a bare hostname or an FQDN (``host.domain``).
|
||
When an FQDN is passed the domain part is also updated.
|
||
|
||
Only sends the fields that need to change — never overwrites the full
|
||
general config object to avoid accidental data loss.
|
||
|
||
Requires OPNsense 22.x or later with the general settings REST API.
|
||
Raises NotImplementedError on older versions that lack this endpoint.
|
||
"""
|
||
if "." in new_hostname:
|
||
hostname, domain = new_hostname.split(".", 1)
|
||
else:
|
||
hostname = new_hostname
|
||
domain = None
|
||
|
||
payload: dict = {"general": {"hostname": hostname}}
|
||
if domain:
|
||
payload["general"]["domain"] = domain
|
||
|
||
try:
|
||
result = self._post("/api/core/general/set", payload)
|
||
except Exception as exc:
|
||
raise NotImplementedError(
|
||
f"OPNsense at {self.hostname} does not support hostname management "
|
||
"via REST API (requires OPNsense 22.x+). "
|
||
"Please set the hostname manually via System → Settings → General "
|
||
f"in the web UI. (Detail: {exc})"
|
||
) from exc
|
||
|
||
logger.info("OPNsense set_hostname → %s (domain: %s): %s", hostname, domain, result)
|
||
|
||
# ── 3. Regenerate web GUI TLS certificate ──
|
||
self._regenerate_web_cert(hostname, domain)
|
||
|
||
def _regenerate_web_cert(self, hostname: str, domain: str) -> None:
|
||
"""Attempt to regenerate the OPNsense web GUI self-signed certificate.
|
||
|
||
Tries the configd helper first (OPNsense 22.x+), then falls back to the
|
||
Trust API (older releases). Logs a warning if neither works — the
|
||
operator should then regenerate the cert manually in the web GUI.
|
||
"""
|
||
fqdn = f"{hostname}.{domain}" if domain else hostname
|
||
|
||
# Attempt 1: configd helper (most common on 22.x / 23.x / 24.x)
|
||
try:
|
||
self._post("/api/core/configd/generateRootCert", {})
|
||
logger.info("OPNsense web cert regenerated via configd (fqdn: %s)", fqdn)
|
||
return
|
||
except Exception as exc:
|
||
logger.debug("configd/generateRootCert not available: %s", exc)
|
||
|
||
# Attempt 2: Trust API — create a new internal self-signed cert
|
||
try:
|
||
# Fetch the list of internal CAs to find the right one
|
||
ca_data = self._post("/api/trust/ca/search", {})
|
||
internal_ca_uuid = None
|
||
for row in (ca_data.get("rows") or []):
|
||
if row.get("internal") == "1" or row.get("catype") == "internal":
|
||
internal_ca_uuid = row.get("uuid")
|
||
break
|
||
|
||
cert_payload: dict = {
|
||
"cert": {
|
||
"descr": f"netork-generated for {fqdn}",
|
||
"caref": internal_ca_uuid or "",
|
||
"keytype": "RSA",
|
||
"keylen": "2048",
|
||
"digest_alg": "sha256",
|
||
"lifetime": "825",
|
||
"dn_commonname": fqdn,
|
||
"dn_sans": fqdn,
|
||
}
|
||
}
|
||
result = self._post("/api/trust/cert/generate", cert_payload)
|
||
logger.info("OPNsense web cert generated via Trust API: %s", result)
|
||
return
|
||
except Exception as exc:
|
||
logger.debug("Trust API cert generation failed: %s", exc)
|
||
|
||
logger.warning(
|
||
"OPNsense: could not regenerate web GUI certificate automatically after "
|
||
"hostname change to %s — please regenerate it manually in the web GUI "
|
||
"(System → Trust → Certificates).",
|
||
fqdn,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# DNS host overrides
|
||
# ------------------------------------------------------------------
|
||
|
||
def _detect_active_dns(self) -> str | None:
|
||
"""Return 'unbound' or 'dnsmasq' depending on which service is running."""
|
||
try:
|
||
data = self._get("/api/core/service/search")
|
||
for row in data.get("rows", []):
|
||
name = (row.get("name") or "").lower()
|
||
if name in ("unbound", "dnsmasq") and row.get("running"):
|
||
return name
|
||
except Exception as exc:
|
||
logger.warning("DNS service detection failed: %s", exc)
|
||
return None
|
||
|
||
def get_dns_entries(self) -> list[dict[str, Any]]:
|
||
"""Return DNS host-override entries from the active DNS service.
|
||
|
||
Checks whether Unbound (DNS Resolver) or Dnsmasq (DNS Forwarder) is
|
||
running and returns host overrides from whichever is active. If neither
|
||
is running, returns an empty list.
|
||
|
||
Each entry::
|
||
|
||
{
|
||
"uuid": str, # OPNsense-internal UUID
|
||
"hostname": str, # host part, e.g. "gw"
|
||
"domain": str, # domain part, e.g. "home.example.com"
|
||
"fqdn": str, # hostname.domain
|
||
"ip": str, # IP address
|
||
"record_type": str, # "A" | "AAAA" | "MX" | …
|
||
"description": str,
|
||
"enabled": bool,
|
||
"service": str, # "unbound" | "dnsmasq"
|
||
}
|
||
"""
|
||
service = self._detect_active_dns()
|
||
if service == "unbound":
|
||
return self._get_unbound_host_overrides()
|
||
if service == "dnsmasq":
|
||
return self._get_dnsmasq_host_overrides()
|
||
return []
|
||
|
||
def _get_unbound_host_overrides(self) -> list[dict[str, Any]]:
|
||
result: list[dict[str, Any]] = []
|
||
seen: set[tuple[str, str, str, str]] = set()
|
||
try:
|
||
data = self._post(
|
||
"/api/unbound/settings/searchhostoverride",
|
||
{"current": 1, "rowCount": -1, "searchPhrase": ""},
|
||
)
|
||
for row in data.get("rows", []):
|
||
host = row.get("hostname", "") or ""
|
||
domain = row.get("domain", "") or ""
|
||
fqdn = f"{host}.{domain}" if host and domain else host or domain
|
||
ip = row.get("server", "") or ""
|
||
rr = row.get("rr", "A") or "A"
|
||
# OPNsense searchhostoverride includes alias records alongside parent
|
||
# records; aliases often have identical content but separate UUIDs.
|
||
# Deduplicate by logical key to avoid inflating the zone with copies.
|
||
key = (host.lower(), domain.lower(), ip, rr)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
result.append({
|
||
"uuid": row.get("uuid", ""),
|
||
"hostname": host,
|
||
"domain": domain,
|
||
"fqdn": fqdn,
|
||
"ip": ip,
|
||
"record_type": rr,
|
||
"description": row.get("description", "") or "",
|
||
"enabled": str(row.get("enabled", "1")) == "1",
|
||
"ptrrecord": str(row.get("ptrrecord", "1")) == "1",
|
||
"service": "unbound",
|
||
})
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch Unbound host overrides: %s", exc)
|
||
return sorted(result, key=lambda e: e["fqdn"].lower())
|
||
|
||
def _get_dnsmasq_host_overrides(self) -> list[dict[str, Any]]:
|
||
result: list[dict[str, Any]] = []
|
||
try:
|
||
data = self._post(
|
||
"/api/dnsmasq/settings/searchhostoverride",
|
||
{"current": 1, "rowCount": -1, "searchPhrase": ""},
|
||
)
|
||
for row in data.get("rows", []):
|
||
host = row.get("host", "") or ""
|
||
domain = row.get("domain", "") or ""
|
||
fqdn = f"{host}.{domain}" if host and domain else host or domain
|
||
result.append({
|
||
"uuid": row.get("uuid", ""),
|
||
"hostname": host,
|
||
"domain": domain,
|
||
"fqdn": fqdn,
|
||
"ip": row.get("ip", "") or "",
|
||
"record_type": "A",
|
||
"description": row.get("description", "") or "",
|
||
"enabled": str(row.get("enabled", "1")) == "1",
|
||
"service": "dnsmasq",
|
||
})
|
||
except Exception as exc:
|
||
logger.warning("Failed to fetch Dnsmasq host overrides: %s", exc)
|
||
return sorted(result, key=lambda e: e["fqdn"].lower())
|
||
|
||
def set_dns_entry(self, uuid: str, service: str, data: dict[str, Any]) -> dict[str, Any]:
|
||
"""Update a single DNS host-override entry and reconfigure the service.
|
||
|
||
``data`` keys: hostname, domain, ip, record_type, description, enabled.
|
||
Returns the raw API response from the set call.
|
||
"""
|
||
if service == "unbound":
|
||
payload = {
|
||
"host": {
|
||
"enabled": "1" if data.get("enabled", True) else "0",
|
||
"hostname": data.get("hostname", ""),
|
||
"domain": data.get("domain", ""),
|
||
"rr": data.get("record_type", "A"),
|
||
"server": data.get("ip", ""),
|
||
"description": data.get("description", "") or "",
|
||
"mxprio": "",
|
||
"mx": "",
|
||
}
|
||
}
|
||
resp = self._post(f"/api/unbound/settings/sethostoverride/{uuid}", payload)
|
||
self._post("/api/unbound/service/reconfigure")
|
||
elif service == "dnsmasq":
|
||
payload = {
|
||
"hostoverride": {
|
||
"enabled": "1" if data.get("enabled", True) else "0",
|
||
"host": data.get("hostname", ""),
|
||
"domain": data.get("domain", ""),
|
||
"ip": data.get("ip", ""),
|
||
"description": data.get("description", "") or "",
|
||
}
|
||
}
|
||
resp = self._post(f"/api/dnsmasq/settings/sethostoverride/{uuid}", payload)
|
||
self._post("/api/dnsmasq/service/reconfigure")
|
||
else:
|
||
raise NotImplementedError(f"set_dns_entry not supported for service '{service}'")
|
||
return resp
|
||
|
||
_NETORK_TAG = "[netork]"
|
||
|
||
def sync_dns_zone(self, zone_name: str, records: list[dict[str, Any]]) -> None:
|
||
"""Replace all netork-managed host overrides for *zone_name* with *records*.
|
||
|
||
records items: {"hostname": str, "ip": str, "record_type": str, "enabled": bool}
|
||
Auto-detects whether Unbound or Dnsmasq is active.
|
||
"""
|
||
zone_lower = zone_name.lower()
|
||
if "in-addr.arpa" in zone_lower or "ip6.arpa" in zone_lower:
|
||
raise ValueError(
|
||
f"Refusing to provision reverse zone '{zone_name}' as OPNsense host overrides — "
|
||
"PTR records must not be managed via the host override API."
|
||
)
|
||
|
||
service = self._detect_active_dns()
|
||
if not service:
|
||
raise RuntimeError("No active DNS service (Unbound or Dnsmasq) detected")
|
||
|
||
zone_clean = zone_name.rstrip(".")
|
||
|
||
# Remove existing netork-managed overrides for this zone
|
||
existing = (self._get_unbound_host_overrides() if service == "unbound"
|
||
else self._get_dnsmasq_host_overrides())
|
||
for entry in existing:
|
||
if (entry["domain"].rstrip(".") == zone_clean
|
||
and entry["description"].startswith(self._NETORK_TAG)):
|
||
uid = entry["uuid"]
|
||
if uid:
|
||
if service == "unbound":
|
||
self._post(f"/api/unbound/settings/delhostoverride/{uid}")
|
||
else:
|
||
self._post(f"/api/dnsmasq/settings/delhostoverride/{uid}")
|
||
logger.info("Removed %s host override %s.%s (uuid %s)",
|
||
service, entry["hostname"], zone_clean, uid)
|
||
|
||
# Re-add all enabled A/AAAA records
|
||
added = 0
|
||
for rec in records:
|
||
if not rec.get("enabled", True):
|
||
continue
|
||
rtype = rec.get("record_type", "A")
|
||
if rtype not in ("A", "AAAA"):
|
||
continue
|
||
if service == "unbound":
|
||
self._post("/api/unbound/settings/addhostoverride", {
|
||
"host": {
|
||
"enabled": "1",
|
||
"hostname": rec.get("hostname", ""),
|
||
"domain": zone_clean,
|
||
"rr": rtype,
|
||
"server": rec.get("ip", ""),
|
||
"description": self._NETORK_TAG,
|
||
"ptrrecord": "1",
|
||
"mxprio": "",
|
||
"mx": "",
|
||
}
|
||
})
|
||
else:
|
||
self._post("/api/dnsmasq/settings/addhostoverride", {
|
||
"hostoverride": {
|
||
"enabled": "1",
|
||
"host": rec.get("hostname", ""),
|
||
"domain": zone_clean,
|
||
"ip": rec.get("ip", ""),
|
||
"description": self._NETORK_TAG,
|
||
}
|
||
})
|
||
added += 1
|
||
logger.info("Added %s host override %s.%s → %s",
|
||
service, rec.get("hostname"), zone_clean, rec.get("ip"))
|
||
|
||
if service == "unbound":
|
||
self._post("/api/unbound/service/reconfigure")
|
||
else:
|
||
self._post("/api/dnsmasq/service/reconfigure")
|
||
logger.info("sync_dns_zone %s via %s: %d records provisioned", zone_clean, service, added)
|
||
|
||
def deprovision_dns_zone(self, zone_name: str) -> None:
|
||
"""Remove all netork-managed host overrides for *zone_name*."""
|
||
self.sync_dns_zone(zone_name, [])
|