1391 lines
55 KiB
Python
1391 lines
55 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 socket
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
from requests.exceptions import RequestException
|
||
|
||
from napalm_device_types import FirewallDriver
|
||
from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException
|
||
|
||
|
||
class OPNsenseDriver(FirewallDriver):
|
||
"""NAPALM driver for OPNsense (read-only, REST API)."""
|
||
|
||
VENDOR = "OPNsense"
|
||
|
||
def __init__(
|
||
self,
|
||
hostname: str,
|
||
username: str,
|
||
password: str,
|
||
timeout: int = 60,
|
||
optional_args: Optional[Dict[str, Any]] = 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}"
|
||
self.verify = self.optional_args.get("verify", True)
|
||
self.api_key = self.optional_args.get("api_key") or username
|
||
self.api_secret = self.optional_args.get("api_secret") or password
|
||
|
||
self.session: Optional[requests.Session] = None
|
||
|
||
# Config-management state
|
||
self._candidate_config: Optional[List[Dict[str, Any]]] = 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: Optional[str] = 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: Optional[Dict[str, Any]] = 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 self.hostname
|
||
version = status.get("version") or status.get("product_version") or "unknown"
|
||
|
||
try:
|
||
interface_list = list(self.get_interfaces().keys())
|
||
except Exception:
|
||
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.
|
||
|
||
Calls ``GET /api/interfaces/addresses/export``.
|
||
|
||
Structure::
|
||
|
||
{
|
||
"em0": {
|
||
"ipv4": {"192.0.2.10": {"prefix_length": 24}},
|
||
"ipv6": {"2001:db8::1": {"prefix_length": 64}},
|
||
}
|
||
}
|
||
"""
|
||
data = self._get("/api/interfaces/addresses/export")
|
||
result: Dict[str, Dict[str, Any]] = {}
|
||
|
||
for item in data.get("items", []):
|
||
ifname: str = item["interface"]
|
||
ip: str = item["address"]
|
||
prefix: int = int(item["prefix"])
|
||
family = "ipv6" if ":" in ip else "ipv4"
|
||
|
||
result.setdefault(ifname, {"ipv4": {}, "ipv6": {}})
|
||
result[ifname][family][ip] = {"prefix_length": prefix}
|
||
|
||
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,
|
||
}
|
||
"""
|
||
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) -> 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,
|
||
})
|
||
except ValueError:
|
||
pass
|
||
|
||
for iface in items:
|
||
name: str = iface.get("device") or iface.get("name", "")
|
||
if not name:
|
||
continue
|
||
|
||
# Primary format: flat CIDR strings in addr4/addr6
|
||
addr4: str = iface.get("addr4", "")
|
||
addr6: str = iface.get("addr6", "")
|
||
if addr4:
|
||
_add(name, addr4, "ipv4")
|
||
if addr6:
|
||
_add(name, addr6, "ipv6")
|
||
|
||
# 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")
|
||
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 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:
|
||
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]]] = {}
|
||
|
||
proto_map = {
|
||
"static": "static",
|
||
"ospf": "ospf",
|
||
"bgp": "bgp",
|
||
"rip": "rip",
|
||
"kernel": "connected",
|
||
"connected": "connected",
|
||
"local": "connected",
|
||
}
|
||
|
||
for route in data.get("route", []):
|
||
network = route.get("network") or route.get("destination", "")
|
||
if not network:
|
||
continue
|
||
|
||
if destination and network != destination:
|
||
continue
|
||
|
||
flags = route.get("flags", "").upper()
|
||
proto_raw = route.get("proto", "").lower()
|
||
proto = proto_map.get(proto_raw, proto_raw)
|
||
|
||
if protocol and proto != protocol.lower():
|
||
continue
|
||
|
||
gateway = route.get("gateway") or route.get("nexthop", "")
|
||
iface = route.get("netif") or route.get("interface", "")
|
||
|
||
entry: Dict[str, Any] = {
|
||
"protocol": proto,
|
||
"current_active": "U" in flags,
|
||
"last_active": False,
|
||
"age": -1,
|
||
"next_hop": gateway if gateway not in ("link#", "0.0.0.0", "") else "",
|
||
"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:
|
||
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:
|
||
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:
|
||
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:
|
||
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:
|
||
pass
|
||
|
||
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: Optional[str] = None, config: Optional[str] = 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: Optional[int] = 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) -> Optional[str]:
|
||
"""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:
|
||
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_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:
|
||
pass
|
||
|
||
# 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:
|
||
pass
|
||
|
||
# 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:
|
||
return []
|
||
|
||
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:
|
||
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:
|
||
# 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:
|
||
pass
|
||
|
||
# ── 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:
|
||
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:
|
||
pass
|
||
|
||
# ── 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:
|
||
pass
|
||
|
||
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:
|
||
pass
|
||
|
||
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("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:
|
||
pass
|
||
return []
|
||
|
||
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:
|
||
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:
|
||
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:
|
||
return {"success": False, "output": str(exc)}
|