fix: accept verify_ssl/ssl_verify optional args; add get_firewall_aliases/rules
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
d926218eff
commit
4434d4195f
+110
-1
@@ -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"]))
|
||||
|
||||
Reference in New Issue
Block a user