feat: get_host_firewall, the host's own netfilter rules (NetOrk/netork#1344)
CI / test (3.10) (push) Failing after 15s
CI / test (3.11) (push) Failing after 14s
CI / test (3.12) (push) Failing after 21s
CI / test (3.10) (pull_request) Failing after 15s
CI / test (3.11) (pull_request) Failing after 17s
CI / test (3.12) (pull_request) Failing after 16s

Mixes in napalm-device-types' HostFirewallMixin. The transport runs the shared
read-only command as root on an exec channel: nft prints the whole ruleset as
one line of JSON, longer than the PTY takes reliably, and both nft and
iptables-save need CAP_NET_ADMIN. Requires napalm-device-types 4.21.0;
version 0.13.0.
This commit is contained in:
Christian Manivong committed 2026-10-21 23:10:48 +02:00
1 parent facc081123
commit 19b86d103d
4 files changed
+59 -2

No files matched your search

+45
View File
@@ -1651,6 +1651,51 @@ class TestGetSystemIdentity:
driver._device.send_command.assert_not_called()
# ---------------------------------------------------------------------------
# get_host_firewall (napalm-device-types HostFirewallMixin, NetOrk/netork#1344)
# ---------------------------------------------------------------------------
_FIREWALL = (
"FW_BEGIN\n[iptables-save]\n"
"# Generated by iptables-save v1.8.10 (nf_tables) on Wed Oct 21 14:52:11 2026\n"
"*filter\n:INPUT ACCEPT [0:0]\n-A INPUT -i ens6 -j DROP\nCOMMIT\n__FW_RC=0\nFW_END\n"
)
class TestGetHostFirewall:
"""nft and iptables-save need root; the nft JSON is one long line, so the
command goes on an exec channel, never typed into the PTY netmiko holds."""
def test_it_reads_the_shared_command_as_root_on_an_exec_channel(self, driver):
from napalm_device_types import CommandResult, HostFirewallMixin
assert isinstance(driver, HostFirewallMixin)
calls = []
def run_command(command, *, privileged=False, timeout=60, stdin=None):
calls.append((command, privileged))
return CommandResult(_FIREWALL, "", 0)
driver.run_command = run_command
reading = driver.get_host_firewall()
assert reading["sources"] == ["iptables-nft"]
assert reading["chains"][0]["rules"][0]["verdict"] == "drop"
[(command, privileged)] = calls
assert privileged is True
assert command.startswith("sh -c '")
driver._device.send_command.assert_not_called()
def test_without_root_there_is_no_reading(self, driver):
from napalm_device_types import CommandResult, HostFirewallUnavailable
driver.run_command = lambda *a, **k: CommandResult("", "sudo: a password is required\n", 1)
with pytest.raises((HostFirewallUnavailable, ValueError)):
driver.get_host_firewall()
# ---------------------------------------------------------------------------
# get_virtual_ips (napalm-device-types VirtualIpsMixin)
# ---------------------------------------------------------------------------