Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fffdd95e6a | ||
|
|
e53cc8d402 | ||
|
|
70fc5f7043 | ||
|
|
c8d63e87e4 |
@@ -12,7 +12,7 @@ jobs:
|
|||||||
strategy:
|
strategy:
|
||||||
fail-fast: false
|
fail-fast: false
|
||||||
matrix:
|
matrix:
|
||||||
python-version: ["3.9", "3.10", "3.11", "3.12"]
|
python-version: ["3.10", "3.11", "3.12"]
|
||||||
steps:
|
steps:
|
||||||
- name: Checkout
|
- name: Checkout
|
||||||
uses: actions/checkout@v4
|
uses: actions/checkout@v4
|
||||||
@@ -26,6 +26,9 @@ jobs:
|
|||||||
- name: Install package with dev extras
|
- name: Install package with dev extras
|
||||||
run: |
|
run: |
|
||||||
python -m pip install --upgrade pip
|
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]"
|
python -m pip install -e ".[dev]"
|
||||||
|
|
||||||
- name: Run unit tests
|
- name: Run unit tests
|
||||||
@@ -38,7 +41,8 @@ jobs:
|
|||||||
python -m build
|
python -m build
|
||||||
|
|
||||||
- name: Upload dist artifacts
|
- 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:
|
with:
|
||||||
name: dist-${{ matrix.python-version }}
|
name: dist-${{ matrix.python-version }}
|
||||||
path: dist/*
|
path: dist/*
|
||||||
@@ -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"]))
|
||||||
|
|||||||
+1
-2
@@ -8,7 +8,7 @@ version = "0.1.0"
|
|||||||
description = "NAPALM driver for OPNsense (read-only via REST API)."
|
description = "NAPALM driver for OPNsense (read-only via REST API)."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = { text = "Apache-2.0" }
|
license = { text = "Apache-2.0" }
|
||||||
requires-python = ">=3.9"
|
requires-python = ">=3.10"
|
||||||
authors = [
|
authors = [
|
||||||
{ name = "Christian Manivong" },
|
{ name = "Christian Manivong" },
|
||||||
]
|
]
|
||||||
@@ -16,7 +16,6 @@ classifiers = [
|
|||||||
"Topic :: Utilities",
|
"Topic :: Utilities",
|
||||||
"License :: OSI Approved :: Apache Software License",
|
"License :: OSI Approved :: Apache Software License",
|
||||||
"Programming Language :: Python :: 3",
|
"Programming Language :: Python :: 3",
|
||||||
"Programming Language :: Python :: 3.9",
|
|
||||||
"Programming Language :: Python :: 3.10",
|
"Programming Language :: Python :: 3.10",
|
||||||
"Programming Language :: Python :: 3.11",
|
"Programming Language :: Python :: 3.11",
|
||||||
"Programming Language :: Python :: 3.12",
|
"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