Merge pull request 'feat: get_host_firewall, the host's own netfilter rules (NetOrk/netork#1344)' (#37) from feat/host-firewall-1344 into master

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `get_host_firewall()` (napalm-device-types' `HostFirewallMixin`, NetOrk/netork#1344): the host's own netfilter rules — `nft -j list ruleset`, with the tables iptables manages taken from `iptables-save` — as chains and rules in one shape. Read-only, as root, on an exec channel (the nft JSON is one long line). Requires napalm-device-types 4.21.0; version 0.13.0.
 - `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd service is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, on an exec channel under `sh`, without sudo. Requires napalm-device-types 4.20.0; version 0.12.0.
 
 ### Security

napalm_linux/linux.py

@@ -40,6 +40,7 @@ from napalm_device_types import (
     ContainerEngineMixin,
     EnsurePackageMixin,
     FingerprintRule,
+    HostFirewallMixin,
     HostStatusMixin,
     KernelFactsMixin,
     ListeningSocketsMixin,
@@ -297,6 +298,7 @@ class LinuxDriver(
     KernelFactsMixin,
     ServicePackagesMixin,
     ListeningSocketsMixin,
+    HostFirewallMixin,
     VirtualIpsMixin,
     ServiceClustersMixin,
     SystemIdentityMixin,
@@ -1188,6 +1190,15 @@ class LinuxDriver(
             return self._run_privileged(command, 60)
         return self._send(command, read_timeout=60)
 
+    def _run_host_firewall_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``HostFirewallMixin.get_host_firewall``: an exec channel.
+
+        ``nft -j`` prints the whole ruleset as one line of JSON, longer than the
+        PTY netmiko holds reliably takes. Read-only; root because both nft and
+        iptables-save need ``CAP_NET_ADMIN`` (NetOrk/netork#1344).
+        """
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
     def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
         """The transport for ``VirtualIpsMixin.get_virtual_ips``: an exec channel.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.12.0"
+version = "0.13.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.20.0",
+    "napalm-device-types>=4.21.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -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)
 # ---------------------------------------------------------------------------