diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index efd9a75..3b1ca46 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -72,7 +72,9 @@ class OPNsenseDriver(FirewallDriver): # 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) + # 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 @@ -1521,3 +1523,110 @@ class OPNsenseDriver(FirewallDriver): lines.append("[warn] Could not verify SNMP state via API.") 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: + 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: + 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: + pass + + 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"]))