feat(firewall): report the gateway a filter rule policy-routes to
CI / test (3.10) (push) Successful in 57s
CI / test (3.11) (push) Successful in 35s
CI / test (3.12) (push) Successful in 33s
CI / test (3.10) (pull_request) Successful in 33s
CI / test (3.11) (pull_request) Successful in 31s
CI / test (3.12) (pull_request) Successful in 35s
CI / test (3.10) (push) Successful in 57s
CI / test (3.11) (push) Successful in 35s
CI / test (3.12) (push) Successful in 33s
CI / test (3.10) (pull_request) Successful in 33s
CI / test (3.11) (pull_request) Successful in 31s
CI / test (3.12) (pull_request) Successful in 35s
get_firewall_rules read searchRule but dropped the rule's gateway. A pass rule with a gateway hands what it matches to that gateway, local destinations included, so it does not reach a host on another internal network. Without the field a caller judging reachability reads a rule meant for internet traffic as a hole into every server: on the first real box, "pass UDP IOT -> any" via WAN_GW made 61 hosts look reachable from the IoT segment (netOrk #575). The rule dict gains "gateway", the gateway's name or "". It is an extra field like floating and interface_label; the generic diff compares a fixed field list and ignores it.
This commit is contained in:
@@ -2444,6 +2444,10 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
|
|||||||
* ``floating`` — bool, rule applies across all interfaces
|
* ``floating`` — bool, rule applies across all interfaces
|
||||||
* ``interface_label`` — human-readable interface description
|
* ``interface_label`` — human-readable interface description
|
||||||
* ``is_group`` — bool, interface is an interface group
|
* ``is_group`` — bool, interface is an interface group
|
||||||
|
* ``gateway`` — the gateway a pass rule policy-routes to, or
|
||||||
|
``""``. Such a rule sends what it matches to that gateway, local
|
||||||
|
destinations included, so it does not reach a host on another
|
||||||
|
internal network.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1")
|
resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1")
|
||||||
@@ -2508,6 +2512,7 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
|
|||||||
"log": str(row.get("log", "0")) == "1",
|
"log": str(row.get("log", "0")) == "1",
|
||||||
"enabled": str(row.get("enabled", "1")) == "1",
|
"enabled": str(row.get("enabled", "1")) == "1",
|
||||||
"category": category,
|
"category": category,
|
||||||
|
"gateway": row.get("gateway", "") or "",
|
||||||
})
|
})
|
||||||
|
|
||||||
return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))
|
return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))
|
||||||
|
|||||||
@@ -0,0 +1,78 @@
|
|||||||
|
"""Filter rules as ``get_firewall_rules`` reports them.
|
||||||
|
|
||||||
|
A pass rule with a gateway is policy routing: OPNsense hands what it matches to
|
||||||
|
that gateway, local destinations included. A caller judging which network can
|
||||||
|
reach which host has to know that, or a rule meant for internet traffic reads
|
||||||
|
as a hole into every server (netOrk #575: on the first real box, "pass UDP
|
||||||
|
IOT -> any" via WAN_GW made 61 hosts look reachable from the IoT segment).
|
||||||
|
|
||||||
|
The row shape follows that box's ``/api/firewall/filter/searchRule``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from napalm_opnsense.opnsense import OPNsenseDriver
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def driver():
|
||||||
|
with patch("napalm_opnsense.opnsense.requests.Session"):
|
||||||
|
yield OPNsenseDriver(
|
||||||
|
hostname="opnsense.example.com",
|
||||||
|
username="api_key",
|
||||||
|
password="api_secret",
|
||||||
|
optional_args={"verify": False},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _row(**over) -> dict:
|
||||||
|
"""One rule row as the API returns it; booleans are "0"/"1" strings."""
|
||||||
|
row = {
|
||||||
|
"uuid": "u1",
|
||||||
|
"sequence": "1",
|
||||||
|
"action": "pass",
|
||||||
|
"quick": "1",
|
||||||
|
"interface": "opt3",
|
||||||
|
"%interface": "IOT",
|
||||||
|
"direction": "in",
|
||||||
|
"ipprotocol": "inet",
|
||||||
|
"protocol": "UDP",
|
||||||
|
"%source_net": "HOME_OFFICE_IOT_NET",
|
||||||
|
"%destination_net": "any",
|
||||||
|
"destination_port": "",
|
||||||
|
"description": "reolink_udp_long_state_timeout",
|
||||||
|
"enabled": "1",
|
||||||
|
"gateway": "WAN_GW",
|
||||||
|
}
|
||||||
|
row.update(over)
|
||||||
|
return row
|
||||||
|
|
||||||
|
|
||||||
|
def _rules(driver, *rows):
|
||||||
|
def fake_get(path):
|
||||||
|
return {"rows": list(rows)} if "searchRule" in path else {"rows": []}
|
||||||
|
|
||||||
|
with patch.object(driver, "_get", side_effect=fake_get):
|
||||||
|
return driver.get_firewall_rules()
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_policy_routed_rule_reports_its_gateway(driver):
|
||||||
|
[rule] = _rules(driver, _row(gateway="WAN_GW"))
|
||||||
|
assert rule["gateway"] == "WAN_GW"
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_rule_that_routes_normally_reports_no_gateway(driver):
|
||||||
|
[rule] = _rules(driver, _row(gateway=""))
|
||||||
|
assert rule["gateway"] == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_row_without_the_field_reports_no_gateway(driver):
|
||||||
|
# Older firmware, or a rule type that never carries one.
|
||||||
|
row = _row()
|
||||||
|
del row["gateway"]
|
||||||
|
[rule] = _rules(driver, row)
|
||||||
|
assert rule["gateway"] == ""
|
||||||
Reference in New Issue
Block a user