feat(firewall): implement apply_firewall_rule + commit_firewall_rules
CI / test (3.10) (push) Failing after 7s
CI / test (3.11) (push) Failing after 7s
CI / test (3.12) (push) Failing after 8s
CI / test (3.9) (push) Failing after 7s

OPNsense-specific half of the FirewallDriver diff/apply mechanism added
in napalm-device-types: translates the vendor-neutral rule dict into the
/api/firewall/filter/addRule or setRule/<uuid> payload (string "1"/"0"
booleans, empty interface = floating rule -- same shape as the existing
SNMP self-provisioning rule in _action_fix_snmp), and commit_firewall_rules
reloads the filter via /api/firewall/filter/apply. get_firewall_rules()
already returns compatible field names, no changes needed there.
This commit is contained in:
Christian Manivong
2026-07-20 15:05:53 +02:00
parent d6a0b21dc6
commit 8d3c443159
2 changed files with 129 additions and 0 deletions
+43
View File
@@ -2149,6 +2149,49 @@ class OPNsenseDriver(FirewallDriver):
return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))
def apply_firewall_rule(self, rule: dict[str, Any], *, uuid: str | None = None) -> dict[str, Any]:
"""Create or update a single OPNsense firewall filter rule.
`rule` uses the vendor-neutral field names from
``napalm_device_types.models.FirewallRuleDict`` (see
``FirewallDriver.diff_firewall_rules``/``apply_firewall_ruleset``,
the generic reconciliation engine that calls this method). This is
the OPNsense-specific half: translating those fields into the
``/api/firewall/filter/addRule``/``setRule`` payload shape (string
"1"/"0" booleans, empty ``interface`` means a floating rule — same
payload shape as the SNMP self-provisioning rule in
``_action_fix_snmp``).
"""
payload = {
"rule": {
"enabled": "1" if rule.get("enabled", True) else "0",
"sequence": "1",
"action": rule.get("action", "pass"),
"quick": "1" if rule.get("quick", True) else "0",
"interface": rule.get("interface", "") or "",
"direction": rule.get("direction", "in"),
"ipprotocol": "inet",
"protocol": rule.get("protocol", "any"),
"source_net": rule.get("source_net", "any") or "any",
"source_port": rule.get("source_port", "") or "",
"destination_net": rule.get("destination_net", "any") or "any",
"destination_port": rule.get("destination_port", "") or "",
"log": "1" if rule.get("log", False) else "0",
"floating": "yes" if not rule.get("interface") else "no",
"descr": rule.get("description", ""),
}
}
path = f"/api/firewall/filter/setRule/{uuid}" if uuid else "/api/firewall/filter/addRule"
return self._post(path, payload)
def commit_firewall_rules(self) -> dict[str, Any]:
"""Reload the firewall filter to activate pending rule changes.
Final step after one or more `apply_firewall_rule()` calls — same
as the last step of `_action_fix_snmp`.
"""
return self._post("/api/firewall/filter/apply", {})
# ------------------------------------------------------------------
# Hostname management
# ------------------------------------------------------------------