Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fffdd95e6a | ||
|
|
e53cc8d402 | ||
|
|
70fc5f7043 | ||
|
|
c8d63e87e4 |
@@ -12,7 +12,7 @@ jobs:
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
python-version: ["3.9", "3.10", "3.11", "3.12"]
|
||||
python-version: ["3.10", "3.11", "3.12"]
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
@@ -26,6 +26,9 @@ jobs:
|
||||
- name: Install package with dev extras
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
# napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
|
||||
# pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
|
||||
python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
|
||||
python -m pip install -e ".[dev]"
|
||||
|
||||
- name: Run unit tests
|
||||
@@ -38,7 +41,8 @@ jobs:
|
||||
python -m build
|
||||
|
||||
- name: Upload dist artifacts
|
||||
uses: actions/upload-artifact@v4
|
||||
# v4 refuses to run on Gitea ("not currently supported on GHES").
|
||||
uses: actions/upload-artifact@v3
|
||||
with:
|
||||
name: dist-${{ matrix.python-version }}
|
||||
path: dist/*
|
||||
@@ -2444,6 +2444,10 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
|
||||
* ``floating`` — bool, rule applies across all interfaces
|
||||
* ``interface_label`` — human-readable interface description
|
||||
* ``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:
|
||||
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",
|
||||
"enabled": str(row.get("enabled", "1")) == "1",
|
||||
"category": category,
|
||||
"gateway": row.get("gateway", "") or "",
|
||||
})
|
||||
|
||||
return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))
|
||||
|
||||
+1
-2
@@ -8,7 +8,7 @@ version = "0.1.0"
|
||||
description = "NAPALM driver for OPNsense (read-only via REST API)."
|
||||
readme = "README.md"
|
||||
license = { text = "Apache-2.0" }
|
||||
requires-python = ">=3.9"
|
||||
requires-python = ">=3.10"
|
||||
authors = [
|
||||
{ name = "Christian Manivong" },
|
||||
]
|
||||
@@ -16,7 +16,6 @@ classifiers = [
|
||||
"Topic :: Utilities",
|
||||
"License :: OSI Approved :: Apache Software License",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
|
||||
@@ -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