Christian Manivong
f934cc0cfa
OPNsense has no synchronous ping endpoint. /api/diagnostics/ping is a job API — create, start, read statistics, stop, remove — so a single ping costs five requests and roughly a second of waiting, which makes the generic per-host sweep from napalm-device-types unusable for a whole subnet. The override exploits what the job API does offer instead: jobs are independent and run on the firewall in parallel, and search_jobs reports all of them in one response. A batch (32 by default) is created and started, waited for once, harvested with a single request and then cleaned up, so waiting time per batch is constant rather than linear in hosts. PING_SWEEP_MAX_TARGETS bounds the sweep as a whole — this runs on production firewalls. Two details the API forces: results are polled, because search_jobs signals the running ping with SIGINFO and then parses whatever it has written so far, so the first read of a healthy host can still show zero probes; and the model's root node is read from /get rather than hardcoded, so a rename in a future OPNsense release cannot silently break job creation. Tested against mocked API responses only — no live device is reachable at the moment, so the endpoint shapes come from the OPNsense sources (PingController, scripts/interfaces/ping.py).
2515 lines
107 KiB
Python
2515 lines
107 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
|
||
|
||
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, [])
|