diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 4009791..e97a254 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -37,8 +37,11 @@ from __future__ import annotations import difflib import json +import logging import socket -from typing import Any, Dict, List, Optional +from typing import Any + +logger = logging.getLogger(__name__) import requests from requests.exceptions import RequestException @@ -58,7 +61,7 @@ class OPNsenseDriver(FirewallDriver): username: str, password: str, timeout: int = 60, - optional_args: Optional[Dict[str, Any]] = None, + optional_args: dict[str, Any] | None = None, ) -> None: self.hostname = hostname self.username = username @@ -78,13 +81,13 @@ class OPNsenseDriver(FirewallDriver): 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 + self.session: requests.Session | None = None # Config-management state - self._candidate_config: Optional[List[Dict[str, Any]]] = None + 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: Optional[str] = None + self._pre_commit_backup_id: str | None = None # ------------------------------------------------------------------ # Connection management @@ -112,7 +115,7 @@ class OPNsenseDriver(FirewallDriver): self.session.close() self.session = None - def is_alive(self) -> Dict[str, bool]: + def is_alive(self) -> dict[str, bool]: """Return whether the session is usable. Performs a lightweight socket-level check without sending a full @@ -133,7 +136,7 @@ class OPNsenseDriver(FirewallDriver): # Internal helpers # ------------------------------------------------------------------ - def _get(self, path: str) -> Dict[str, Any]: + def _get(self, path: str) -> dict[str, Any]: """Perform a GET request against the OPNsense REST API. :raises ConnectionClosedException: if called before :meth:`open`. @@ -146,7 +149,7 @@ class OPNsenseDriver(FirewallDriver): response.raise_for_status() return response.json() - def _post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + 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``. @@ -165,7 +168,7 @@ class OPNsenseDriver(FirewallDriver): # NAPALM getters # ------------------------------------------------------------------ - def get_facts(self) -> Dict[str, Any]: + def get_facts(self) -> dict[str, Any]: """Return a dictionary of general device facts. Calls ``GET /api/core/system/status``. @@ -176,12 +179,13 @@ class OPNsenseDriver(FirewallDriver): """ status = self._get("/api/core/system/status") - hostname = status.get("hostname") or status.get("name") or self.hostname + 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: + except Exception as exc: + logger.debug("Failed to fetch interface list: %s", exc) interface_list = [] return { @@ -195,7 +199,7 @@ class OPNsenseDriver(FirewallDriver): "interface_list": interface_list, } - def get_interfaces(self) -> Dict[str, Dict[str, Any]]: + def get_interfaces(self) -> dict[str, dict[str, Any]]: """Return interface details keyed by interface name. Calls ``GET /api/interfaces/overview/export``. @@ -205,7 +209,7 @@ class OPNsenseDriver(FirewallDriver): ``mac_address``, ``speed``, ``mtu``. """ data = self._get("/api/interfaces/overview/export") - interfaces: Dict[str, Dict[str, Any]] = {} + interfaces: dict[str, dict[str, Any]] = {} # API returns a bare list in newer OPNsense versions; # older/wrapped format uses {"interfaces": [...]} @@ -227,7 +231,7 @@ class OPNsenseDriver(FirewallDriver): return interfaces - def get_interfaces_ip(self) -> Dict[str, Dict[str, Any]]: + 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). @@ -245,7 +249,7 @@ class OPNsenseDriver(FirewallDriver): data = self._get("/api/interfaces/overview/export") items: list = data if isinstance(data, list) else data.get("interfaces", []) - result: Dict[str, Dict[str, Any]] = {} + result: dict[str, dict[str, Any]] = {} def _add(ifname: str, cidr: str, family: str) -> None: try: @@ -284,7 +288,7 @@ class OPNsenseDriver(FirewallDriver): return result - def get_networks(self) -> List[Dict[str, Any]]: + 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``). @@ -309,7 +313,7 @@ class OPNsenseDriver(FirewallDriver): data = self._get("/api/interfaces/overview/export") items: list = data if isinstance(data, list) else data.get("interfaces", []) - networks: List[Dict[str, Any]] = [] + 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.""" @@ -368,7 +372,7 @@ class OPNsenseDriver(FirewallDriver): return networks - def get_arp_table(self, vrf: str = "") -> List[Dict[str, Any]]: + def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]: """Return the ARP table. Calls ``GET /api/diagnostics/interface/get_arp``. @@ -376,7 +380,7 @@ class OPNsenseDriver(FirewallDriver): Each entry contains: ``interface``, ``mac``, ``ip``, ``age``. """ data = self._get("/api/diagnostics/interface/get_arp") - arp_table: List[Dict[str, Any]] = [] + arp_table: list[dict[str, Any]] = [] for entry in data if isinstance(data, list) else data.get("arp", []): arp_table.append( @@ -390,7 +394,7 @@ class OPNsenseDriver(FirewallDriver): return arp_table - def get_interfaces_counters(self) -> Dict[str, Dict[str, Any]]: + 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``. @@ -403,7 +407,7 @@ class OPNsenseDriver(FirewallDriver): ``tx_broadcast_packets``, ``rx_broadcast_packets``. """ data = self._get("/api/diagnostics/interface/get_interface_statistics") - counters: Dict[str, Dict[str, Any]] = {} + counters: dict[str, dict[str, Any]] = {} for iface, stats in data.get("statistics", {}).items(): counters[iface] = { @@ -423,7 +427,7 @@ class OPNsenseDriver(FirewallDriver): return counters - def get_environment(self) -> Dict[str, Any]: + def get_environment(self) -> dict[str, Any]: """Return device environment data (CPU, memory, temperature). Calls: @@ -441,10 +445,11 @@ class OPNsenseDriver(FirewallDriver): try: temp_data = self._get("/api/diagnostics/system/system_temperature") - except Exception: + except Exception as exc: + logger.debug("Failed to fetch system temperature: %s", exc) temp_data = {} - temperature: Dict[str, Any] = {} + 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)) @@ -470,7 +475,7 @@ class OPNsenseDriver(FirewallDriver): destination: str = "", protocol: str = "", longer: bool = False, - ) -> Dict[str, List[Dict[str, Any]]]: + ) -> dict[str, list[dict[str, Any]]]: """Return routing table entries. Calls ``GET /api/diagnostics/interface/get_routes``. @@ -482,7 +487,7 @@ class OPNsenseDriver(FirewallDriver): 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]]] = {} + routes: dict[str, list[dict[str, Any]]] = {} # Optionally enrich with OSPF routes from FRR/Quagga ospf_networks: set = set() @@ -490,8 +495,8 @@ class OPNsenseDriver(FirewallDriver): 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: - pass + 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: @@ -526,7 +531,7 @@ class OPNsenseDriver(FirewallDriver): if protocol and proto != protocol.lower(): continue - entry: Dict[str, Any] = { + entry: dict[str, Any] = { "protocol": proto, "family": family, "current_active": "U" in flags, @@ -544,7 +549,7 @@ class OPNsenseDriver(FirewallDriver): return routes - def get_ipv6_neighbors_table(self) -> List[Dict[str, Any]]: + def get_ipv6_neighbors_table(self) -> list[dict[str, Any]]: """Return the IPv6 Neighbor Discovery (NDP) table. Calls ``GET /api/diagnostics/interface/get_ndp``. @@ -553,7 +558,7 @@ class OPNsenseDriver(FirewallDriver): ``state`` (best-effort from NDP flags). """ data = self._get("/api/diagnostics/interface/get_ndp") - neighbors: List[Dict[str, Any]] = [] + neighbors: list[dict[str, Any]] = [] rows = data if isinstance(data, list) else data.get("rows", []) for entry in rows: @@ -569,7 +574,7 @@ class OPNsenseDriver(FirewallDriver): return neighbors - def get_lldp_neighbors(self) -> Dict[str, List[Dict[str, Any]]]: + def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]: """Return LLDP neighbors grouped by local port. Calls ``GET /api/lldpd/service/neighbor``. @@ -579,10 +584,11 @@ class OPNsenseDriver(FirewallDriver): """ try: data = self._get("/api/lldpd/service/neighbor") - except Exception: + except Exception as exc: + logger.debug("LLDP neighbor fetch failed (plugin not installed?): %s", exc) return {} - neighbors: Dict[str, List[Dict[str, Any]]] = {} + 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( @@ -595,7 +601,7 @@ class OPNsenseDriver(FirewallDriver): def get_lldp_neighbors_detail( self, interface: str = "" - ) -> Dict[str, List[Dict[str, Any]]]: + ) -> dict[str, list[dict[str, Any]]]: """Return detailed LLDP neighbor information. Calls ``GET /api/lldpd/service/neighbor``. @@ -607,10 +613,11 @@ class OPNsenseDriver(FirewallDriver): """ try: data = self._get("/api/lldpd/service/neighbor") - except Exception: + except Exception as exc: + logger.debug("LLDP neighbor detail fetch failed (plugin not installed?): %s", exc) return {} - details: Dict[str, List[Dict[str, Any]]] = {} + 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: @@ -637,7 +644,7 @@ class OPNsenseDriver(FirewallDriver): ) return details - def get_ntp_servers(self) -> Dict[str, Dict[str, Any]]: + 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 @@ -648,17 +655,18 @@ class OPNsenseDriver(FirewallDriver): """ try: data = self._get("/api/ntpd/service/status") - except Exception: + except Exception as exc: + logger.debug("NTP status fetch failed (plugin not installed?): %s", exc) return {} - servers: Dict[str, Dict[str, Any]] = {} + 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]]: + def get_vlans(self) -> dict[str, dict[str, Any]]: """Return configured VLANs. Calls ``GET /api/interfaces/vlan_settings/search_item``. @@ -677,7 +685,7 @@ class OPNsenseDriver(FirewallDriver): ``interfaces`` (list with the VLAN device name). """ data = self._get("/api/interfaces/vlan_settings/search_item") - vlans: Dict[str, Dict[str, Any]] = {} + vlans: dict[str, dict[str, Any]] = {} for row in data.get("rows", []): tag = str(row.get("tag", "")).strip() @@ -696,7 +704,7 @@ class OPNsenseDriver(FirewallDriver): return vlans - def get_bgp_neighbors(self) -> Dict[str, Any]: + def get_bgp_neighbors(self) -> dict[str, Any]: """Return BGP neighbor state. Requires the FRR plugin (``os-frr``) to be installed on OPNsense. @@ -720,7 +728,8 @@ class OPNsenseDriver(FirewallDriver): try: bgp_cfg = self._get("/api/quagga/bgp/get") neighbors_data = self._get("/api/quagga/diagnostics/bgpneighbors") - except Exception: + except Exception as exc: + logger.debug("BGP config fetch failed (plugin not installed?): %s", exc) return {} bgp = bgp_cfg.get("bgp", {}) @@ -737,7 +746,7 @@ class OPNsenseDriver(FirewallDriver): "ipv6Unicast": "ipv6", } - peers: Dict[str, Any] = {} + peers: dict[str, Any] = {} for peer_ip, nbr in raw_neighbors.items(): if not isinstance(nbr, dict): continue @@ -746,7 +755,7 @@ class OPNsenseDriver(FirewallDriver): uptime_msec = int(nbr.get("bgpTimerUpMsec", 0) or 0) uptime = uptime_msec // 1000 if is_up else -1 - address_family: Dict[str, Any] = {} + 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: @@ -788,7 +797,7 @@ class OPNsenseDriver(FirewallDriver): full: bool = False, sanitized: bool = False, format: str = "text", - ) -> Dict[str, str]: + ) -> dict[str, str]: """Return device configuration. OPNsense stores its configuration as XML. This getter returns the @@ -799,7 +808,7 @@ class OPNsenseDriver(FirewallDriver): Calls ``GET /api/core/backup/download/this``. """ - configs: Dict[str, str] = {"running": "", "startup": "", "candidate": ""} + configs: dict[str, str] = {"running": "", "startup": "", "candidate": ""} if retrieve in ("all", "running", "startup"): try: @@ -811,8 +820,8 @@ class OPNsenseDriver(FirewallDriver): configs["running"] = xml_text if retrieve in ("all", "startup"): configs["startup"] = xml_text - except Exception: - pass + 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) @@ -823,7 +832,7 @@ class OPNsenseDriver(FirewallDriver): # Config management (static routes) # ------------------------------------------------------------------ - def load_merge_candidate(self, filename: Optional[str] = None, config: Optional[str] = None) -> None: + 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. @@ -907,7 +916,7 @@ class OPNsenseDriver(FirewallDriver): ) return "".join(diff) - def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None: + 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 @@ -977,7 +986,7 @@ class OPNsenseDriver(FirewallDriver): # Internal helpers (config management) # ------------------------------------------------------------------ - def _get_latest_backup_id(self) -> Optional[str]: + 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 @@ -988,17 +997,18 @@ class OPNsenseDriver(FirewallDriver): data = self._get("/api/core/backup/backups/this") items = data.get("items", []) return items[0]["id"] if items else None - except Exception: + except Exception as exc: + logger.debug("No backup snapshots found: %s", exc) return None - def _fetch_current_routes(self) -> List[Dict[str, Any]]: + 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]] = [] + routes: list[dict[str, Any]] = [] for row in data.get("rows", []): routes.append( { @@ -1014,7 +1024,7 @@ class OPNsenseDriver(FirewallDriver): # NetOrch extensions: packages, services, updates # ------------------------------------------------------------------ - def get_packages(self) -> List[Dict[str, Any]]: + def get_packages(self) -> list[dict[str, Any]]: """Return installed OPNsense plugins. Calls ``GET /api/core/firmware/info`` and returns the ``plugin`` @@ -1024,7 +1034,7 @@ class OPNsenseDriver(FirewallDriver): ``{name, version, installed, description, size, source}`` """ info = self._get("/api/core/firmware/info") - result: List[Dict[str, Any]] = [] + result: list[dict[str, Any]] = [] for p in info.get("plugin", []): if p.get("installed") != "1": continue @@ -1038,7 +1048,7 @@ class OPNsenseDriver(FirewallDriver): }) return sorted(result, key=lambda x: x["name"].lower()) - def get_dhcp_leases(self) -> List[Dict[str, Any]]: + def get_dhcp_leases(self) -> list[dict[str, Any]]: """Return active DHCP leases from OPNsense. Tries all known DHCP backends in order: @@ -1058,7 +1068,7 @@ class OPNsenseDriver(FirewallDriver): * ``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]]: + 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" @@ -1088,8 +1098,8 @@ class OPNsenseDriver(FirewallDriver): leases = _parse_kea(rows) if leases: return leases - except Exception: - pass + except Exception as exc: + logger.debug("Kea DHCP lease fetch failed: %s", exc) # 2. ISC DHCP (legacy) try: @@ -1100,8 +1110,8 @@ class OPNsenseDriver(FirewallDriver): leases = _parse_kea(data.get("rows", [])) if leases: return leases - except Exception: - pass + except Exception as exc: + logger.debug("ISC DHCP lease fetch failed: %s", exc) # 3. ARP table fallback (IP only, no hostname) try: @@ -1111,10 +1121,11 @@ class OPNsenseDriver(FirewallDriver): for e in arp if e.get("mac") and e.get("ip") ] - except Exception: + except Exception as exc: + logger.warning("ARP table fallback failed: %s", exc) return [] - def get_services(self) -> List[Dict[str, Any]]: + def get_services(self) -> list[dict[str, Any]]: """Return running services from OPNsense. Calls ``GET /api/core/service/search`` and normalises the rows to @@ -1122,7 +1133,7 @@ class OPNsenseDriver(FirewallDriver): OpenWrt driver so the UI can render them identically. """ data = self._get("/api/core/service/search") - result: List[Dict[str, Any]] = [] + result: list[dict[str, Any]] = [] for row in data.get("rows", []): result.append({ "name": row.get("name") or row.get("id", ""), @@ -1132,7 +1143,7 @@ class OPNsenseDriver(FirewallDriver): }) return sorted(result, key=lambda x: x["name"].lower()) - def manage_service(self, name: str, action: str) -> Dict[str, Any]: + 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 @@ -1152,9 +1163,10 @@ class OPNsenseDriver(FirewallDriver): 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]]: + 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 @@ -1171,7 +1183,7 @@ class OPNsenseDriver(FirewallDriver): Instance status obtained from ``GET /api/openvpn/service/show``. * **WireGuard** — ``GET /api/wireguard/service/show`` """ - tunnels: Dict[str, Dict[str, Any]] = {} + tunnels: dict[str, dict[str, Any]] = {} # ── IPsec ────────────────────────────────────────────────────────── try: @@ -1233,7 +1245,8 @@ class OPNsenseDriver(FirewallDriver): "bytes_out": bytes_out, "description": description, } - except Exception: + 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( @@ -1255,8 +1268,8 @@ class OPNsenseDriver(FirewallDriver): "bytes_out": int(row.get("bytes-out", 0) or 0), "description": row.get("con", ""), } - except Exception: - pass + except Exception as exc: + logger.debug("Legacy IPsec Phase-2 lease fetch failed: %s", exc) # ── OpenVPN ──────────────────────────────────────────────────────── try: @@ -1266,8 +1279,9 @@ class OPNsenseDriver(FirewallDriver): 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: 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)}" @@ -1287,8 +1301,8 @@ class OPNsenseDriver(FirewallDriver): "bytes_out": 0, "description": name, } - except Exception: - pass + except Exception as exc: + logger.debug("OpenVPN instance fetch failed (plugin not installed?): %s", exc) # ── WireGuard ────────────────────────────────────────────────────── try: @@ -1337,12 +1351,12 @@ class OPNsenseDriver(FirewallDriver): "bytes_out": bytes_out, "description": description, } - except Exception: - pass + 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]]: + def get_available_updates(self) -> list[dict[str, Any]]: """Return available firmware and package updates. Triggers an async update-check on OPNsense via @@ -1355,8 +1369,8 @@ class OPNsenseDriver(FirewallDriver): import time try: self._post("/api/core/firmware/check") - except Exception: - pass + except Exception as exc: + logger.debug("Firmware update check trigger failed: %s", exc) for _ in range(5): time.sleep(3) @@ -1379,17 +1393,17 @@ class OPNsenseDriver(FirewallDriver): ] if state == "latest": return [] - except Exception: - pass + except Exception as exc: + logger.debug("Firmware status poll failed: %s", exc) return [] - def get_device_warnings(self) -> List[Dict[str, Any]]: + 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]] = [] + warnings: list[dict[str, Any]] = [] try: status = self._get("/api/core/firmware/status") state = status.get("status", "none") @@ -1409,11 +1423,11 @@ class OPNsenseDriver(FirewallDriver): "packages": [u.get("name", "") for u in upgrades[:10]], }, }) - except Exception: - pass + 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]: + 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 @@ -1427,9 +1441,10 @@ class OPNsenseDriver(FirewallDriver): 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]]: + 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`` @@ -1439,7 +1454,7 @@ class OPNsenseDriver(FirewallDriver): """ q = query.lower() info = self._get("/api/core/firmware/info") - result: List[Dict[str, Any]] = [] + result: list[dict[str, Any]] = [] for p in info.get("plugin", []): name = p.get("name", "") comment = p.get("comment", "") @@ -1454,7 +1469,7 @@ class OPNsenseDriver(FirewallDriver): }) return sorted(result, key=lambda x: x["name"].lower()) - def install_package(self, name: str) -> Dict[str, Any]: + def install_package(self, name: str) -> dict[str, Any]: """Install an OPNsense plugin by name. Calls ``POST /api/core/firmware/install/{name}``. @@ -1466,9 +1481,10 @@ class OPNsenseDriver(FirewallDriver): 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]: + def uninstall_package(self, name: str) -> dict[str, Any]: """Remove an OPNsense plugin by name. Calls ``POST /api/core/firmware/remove/{name}``. @@ -1480,6 +1496,7 @@ class OPNsenseDriver(FirewallDriver): 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 ────────────────────────────────────────────────────────── @@ -1502,7 +1519,8 @@ class OPNsenseDriver(FirewallDriver): if response.status_code != 200: return None data = response.json() - except Exception: + except Exception as exc: + logger.debug("SNMP config fetch failed (plugin not installed?): %s", exc) return None if not data: @@ -1516,13 +1534,13 @@ class OPNsenseDriver(FirewallDriver): community = general.get("community", "public") or "public" return SNMPConfigDict(running=True, community=community, port=161, version="2c") - def run_device_action(self, action: str) -> Dict[str, Any]: + 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]: + def _action_fix_snmp(self) -> dict[str, Any]: """Install os-net-snmp plugin, configure community 'public', start service. Steps: @@ -1537,6 +1555,7 @@ class OPNsenseDriver(FirewallDriver): result = self._post("/api/core/firmware/install/os-net-snmp") lines.append(f"[install] {result}") except Exception as exc: + logger.debug("os-net-snmp install skipped or failed: %s", exc) lines.append(f"[install] skipped or already installed: {exc}") # 2. Configure SNMP: enable + set community 'public' @@ -1553,25 +1572,29 @@ class OPNsenseDriver(FirewallDriver): }) lines.append("[config] SNMP enabled with community 'public'.") 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: + 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 exc: - lines.append(f"[service] start failed: {exc}") + except Exception as exc2: + logger.debug("SNMP service start failed: %s", exc2) + lines.append(f"[service] start failed: {exc2}") # 4. Verify 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")) - except Exception: + except Exception as exc: + logger.debug("SNMP verification failed: %s", exc) success = False if success: @@ -1581,7 +1604,7 @@ class OPNsenseDriver(FirewallDriver): return {"success": success, "output": "\n".join(lines)} - def get_firewall_aliases(self) -> List[Dict[str, Any]]: + def get_firewall_aliases(self) -> list[dict[str, Any]]: """Return all firewall aliases, sorted by type then name. Each entry contains: @@ -1594,7 +1617,8 @@ class OPNsenseDriver(FirewallDriver): """ try: resp = self._get("/api/firewall/alias/searchItem?current=1&rowCount=-1") - except Exception: + except Exception as exc: + logger.warning("Failed to fetch firewall aliases: %s", exc) return [] rows = resp.get("rows") or [] @@ -1614,7 +1638,7 @@ class OPNsenseDriver(FirewallDriver): return sorted(result, key=lambda x: (x["type"], x["name"].lower())) - def get_firewall_rules(self) -> List[Dict[str, Any]]: + def get_firewall_rules(self) -> list[dict[str, Any]]: """Return all firewall filter rules with interface labels. Extra fields beyond NAPALM standard: @@ -1624,7 +1648,8 @@ class OPNsenseDriver(FirewallDriver): """ try: resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1") - except Exception: + except Exception as exc: + logger.warning("Failed to fetch firewall rules: %s", exc) return [] # OPNsense provides all human-readable values via %-prefixed fields — @@ -1638,8 +1663,8 @@ class OPNsenseDriver(FirewallDriver): n = g.get("ifname") or g.get("name") or "" if n: group_names.add(n) - except Exception: - pass + except Exception as exc: + logger.warning("Failed to fetch interface groups: %s", exc) rows = resp.get("rows") or [] result = [] @@ -1687,3 +1712,322 @@ class OPNsenseDriver(FirewallDriver): }) return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"])) + + # ------------------------------------------------------------------ + # 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, [])