feat: HostFirewallMixin, a Linux host's own netfilter rules (NetOrk/netork#1344)

get_host_firewall() reads `nft -j list ruleset` as root and returns every
chain with its family, table, hook, priority and policy, and every rule as
matches and a verdict. Rules iptables-nft wrote show their extensions in nft's
JSON only as an opaque `xt`, so the tables iptables manages are taken from
iptables-save instead; a legacy iptables is kept beside nft's tables, since
both apply. A condition the parse does not model is an `unknown` match with
its text, never left out. Without root neither tool can be read, and
HostFirewallUnavailable is raised rather than a host without rules.

A driver supplies only _run_host_firewall_command(command, privileged=).
Version 4.21.0.

napalm_device_types/__init__.py

@@ -47,6 +47,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
 * :class:`~napalm_device_types.kernel.KernelFactsMixin`
 * :class:`~napalm_device_types.listening.ListeningSocketsMixin`
+* :class:`~napalm_device_types.host_firewall.HostFirewallMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
@@ -105,6 +106,11 @@ from napalm_device_types.service_packages import (
     parse_service_packages,
 )
 from napalm_device_types.lag import add_lag_interfaces
+from napalm_device_types.host_firewall import (
+    HostFirewallMixin,
+    HostFirewallUnavailable,
+    parse_host_firewall,
+)
 from napalm_device_types.listening import (
     LISTENING_SOCKETS_COMMAND,
     ListeningSocketsMixin,
@@ -275,6 +281,9 @@ __all__ = [
     "ServicePackagesMixin",
     "parse_service_packages",
     "LISTENING_SOCKETS_COMMAND",
+    "HostFirewallMixin",
+    "HostFirewallUnavailable",
+    "parse_host_firewall",
     "ListeningSocketsMixin",
     "ListeningSocketsUnavailable",
     "parse_listening_sockets",

napalm_device_types/host_firewall.py

@@ -0,0 +1,764 @@
+# -*- coding: utf-8 -*-
+"""A Linux host's own firewall: its netfilter rules, read once, in one shape.
+
+Whether a service is reachable from outside its host is decided twice: by the
+address it listens on (:mod:`napalm_device_types.listening`), and by the host's
+own firewall, which may drop the packets first. Reading the firewall is the
+same on every Linux host, so the command and its parse live here, and a driver
+only carries the command across.
+
+**What is read.** ``nft -j list ruleset`` is the primary reading: every table,
+the host's own ``nftables.service`` ones included. Rules that iptables-nft wrote
+are in it too, but their extensions show only as an opaque ``xt`` -- a
+``-m conntrack --ctstate`` or ``-m set`` says nothing there. ``iptables-save``
+spells them out, so the tables iptables manages are taken from it, and the
+others from nft. A *legacy* iptables is a second ruleset beside nft, and both
+apply to every packet, so then both are kept whole.
+
+**One shape.** Each chain says its family, table, hook, priority and policy;
+each rule a list of matches and a verdict. A match is ``field``, ``op``
+(``==``/``!=``) and ``values``. The fields are the few that decide reachability
+from outside: ``iifname``, ``oifname``, ``saddr``, ``daddr``, ``l4proto``,
+``dport``, ``sport``, ``ct_state``, ``ct_orig_dport`` and ``nfproto``.
+Everything else -- an ipset, ``addrtype``, a rate limit, a ``fib`` lookup -- is
+``unknown`` with its text, never dropped: a rule that matches on something this
+does not model must not read as one that matches everything.
+
+**Root, and only root.** Both tools need ``CAP_NET_ADMIN``; without it they
+print an error, not an empty ruleset. So there is no reading without root:
+:class:`HostFirewallUnavailable` is raised rather than a host without rules.
+
+Whether a rule set makes a service unreachable is the consumer's call.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import json
+import re
+import shlex
+from shlex import quote
+from typing import Any, Dict, Iterator, List, Optional, Set, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    HostFirewallDict,
+    NetfilterChainDict,
+    NetfilterMatchDict,
+    NetfilterRuleDict,
+)
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "FW_BEGIN"
+_END = "FW_END"
+_RC_RE = re.compile(r"^__FW_RC=(\d+)$")
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+#: Both tools live in ``/usr/sbin`` on most systems, outside a login user's ``PATH``.
+HOST_FIREWALL_COMMAND = (
+    "PATH=$PATH:/usr/sbin:/sbin; "
+    "printf '%s%s\\n' FW_ BEGIN; "
+    "if command -v nft >/dev/null 2>&1; then "
+    "echo '[nft]'; nft -j list ruleset 2>&1; echo \"__FW_RC=$?\"; fi; "
+    "for t in iptables-save ip6tables-save; do "
+    "if command -v $t >/dev/null 2>&1; then "
+    'echo "[$t]"; $t 2>&1; echo "__FW_RC=$?"; fi; done; '
+    "printf '%s%s\\n' FW_ END"
+)
+
+#: ``iptables-save`` reads the IPv4 tables, ``ip6tables-save`` the IPv6 ones.
+_IPTABLES_FAMILY = {"iptables-save": "ip", "ip6tables-save": "ip6"}
+
+#: The built-in chains of iptables' tables: ``(type, hook, priority)``, as
+#: iptables-nft registers them (and as the legacy tables run in the kernel).
+_BUILT_IN: Dict[str, Dict[str, Tuple[str, str, int]]] = {
+    "raw": {"PREROUTING": ("filter", "prerouting", -300), "OUTPUT": ("filter", "output", -300)},
+    "mangle": {
+        "PREROUTING": ("filter", "prerouting", -150),
+        "INPUT": ("filter", "input", -150),
+        "FORWARD": ("filter", "forward", -150),
+        "OUTPUT": ("route", "output", -150),
+        "POSTROUTING": ("filter", "postrouting", -150),
+    },
+    "nat": {
+        "PREROUTING": ("nat", "prerouting", -100),
+        "INPUT": ("nat", "input", 100),
+        "OUTPUT": ("nat", "output", -100),
+        "POSTROUTING": ("nat", "postrouting", 100),
+    },
+    "filter": {
+        "INPUT": ("filter", "input", 0),
+        "FORWARD": ("filter", "forward", 0),
+        "OUTPUT": ("filter", "output", 0),
+    },
+    "security": {
+        "INPUT": ("filter", "input", 50),
+        "FORWARD": ("filter", "forward", 50),
+        "OUTPUT": ("filter", "output", 50),
+    },
+}
+
+_TERMINAL = {"ACCEPT": "accept", "DROP": "drop", "REJECT": "reject", "RETURN": "return"}
+#: Targets that look at a packet, mark it or log it, and let it go on.
+_NON_TERMINAL = frozenset(
+    {
+        "LOG",
+        "NFLOG",
+        "ULOG",
+        "TRACE",
+        "MARK",
+        "CONNMARK",
+        "CT",
+        "NOTRACK",
+        "TCPMSS",
+        "CHECKSUM",
+        "AUDIT",
+        "SET",
+        "DSCP",
+        "TOS",
+        "TTL",
+        "HL",
+        "CLASSIFY",
+        "CONNSECMARK",
+        "SECMARK",
+        "IDLETIMER",
+        "LED",
+        "RATEEST",
+    }
+)
+
+
+class HostFirewallUnavailable(NotImplementedError):
+    """Nothing on the host could be read: no ``nft``, no ``iptables-save``, or no root."""
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.rstrip() for line in strip_terminal_codes(output).splitlines()]
+    stripped = [line.strip() for line in lines]
+    try:
+        start = stripped.index(_BEGIN)
+        end = stripped.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact host firewall report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, Tuple[List[str], Optional[int]]]:
+    """Each tool's output and exit status."""
+    sections: Dict[str, Tuple[List[str], Optional[int]]] = {}
+    name: Optional[str] = None
+    body: List[str] = []
+    for line in lines:
+        text = line.strip()
+        if text in ("[nft]", "[iptables-save]", "[ip6tables-save]"):
+            name, body = text[1:-1], []
+            sections[name] = (body, None)
+            continue
+        rc = _RC_RE.match(text)
+        if rc and name is not None:
+            sections[name] = (body, int(rc.group(1)))
+            name = None
+            continue
+        if name is not None:
+            body.append(line)
+    return sections
+
+
+# ── iptables-save ────────────────────────────────────────────────────────────
+
+#: Options that start a new part of a rule, outside any ``-m`` module.
+_TOP_LEVEL = frozenset(
+    {
+        "-m",
+        "--match",
+        "-j",
+        "--jump",
+        "-g",
+        "--goto",
+        "-p",
+        "--protocol",
+        "-s",
+        "--source",
+        "-d",
+        "--destination",
+        "-i",
+        "--in-interface",
+        "-o",
+        "--out-interface",
+        "-f",
+        "--fragment",
+        "-c",
+        "--set-counters",
+    }
+)
+#: Modules whose options this reads; any other module is an ``unknown`` match.
+_KNOWN_MODULES = frozenset(
+    {"tcp", "udp", "sctp", "udplite", "multiport", "state", "conntrack", "comment"}
+)
+
+
+def _match(field: str, negated: bool, values: List[str]) -> NetfilterMatchDict:
+    return {"field": field, "op": "!=" if negated else "==", "values": values}
+
+
+def _unknown(text: str) -> NetfilterMatchDict:
+    return {"field": "unknown", "op": "==", "values": [text]}
+
+
+def _iface(name: str) -> str:
+    """iptables writes a wildcard interface as ``eth+``; nft as ``eth*``."""
+    return name[:-1] + "*" if name.endswith("+") else name
+
+
+def _ports(spec: str) -> List[str]:
+    return [p.replace(":", "-") for p in spec.split(",") if p]
+
+
+def _states(spec: str) -> List[str]:
+    return [s.lower() for s in spec.split(",") if s]
+
+
+class _RuleTokens:
+    """The tokens of one ``-A`` line, read front to back."""
+
+    def __init__(self, tokens: List[str]):
+        self.tokens = tokens
+        self.i = 0
+
+    def more(self) -> bool:
+        return self.i < len(self.tokens)
+
+    def peek(self, ahead: int = 0) -> Optional[str]:
+        j = self.i + ahead
+        return self.tokens[j] if j < len(self.tokens) else None
+
+    def take(self) -> str:
+        token = self.tokens[self.i]
+        self.i += 1
+        return token
+
+
+def _module_options(module: str, rt: _RuleTokens, matches: List[NetfilterMatchDict]) -> None:
+    """Read *module*'s options from *rt* into *matches*; stop at the next top-level option."""
+    text = ["-m", module]
+    unknown = module not in _KNOWN_MODULES
+    while rt.more():
+        token = rt.peek()
+        if token in _TOP_LEVEL or (token == "!" and rt.peek(1) in _TOP_LEVEL):
+            break
+        negated = False
+        if token == "!":
+            rt.take()
+            negated = True
+            text.append("!")
+        option = rt.take()
+        value = rt.take() if rt.more() and not str(rt.peek()).startswith("-") else ""
+        text += [option] + ([value] if value else [])
+        if unknown:
+            continue
+        mapped = _module_match(module, option, value, negated)
+        if mapped is None:
+            unknown = True
+        elif mapped is not False:
+            matches.append(mapped)
+    if unknown and module != "comment":
+        matches.append(_unknown(" ".join(text)))
+
+
+def _module_match(module: str, option: str, value: str, negated: bool) -> Any:
+    """The match one module option makes; False for one that matches nothing new; None if not read."""
+    if module == "comment":
+        return False
+    if module in ("tcp", "udp", "sctp", "udplite"):
+        if option in ("--dport", "--destination-port"):
+            return _match("dport", negated, _ports(value))
+        if option in ("--sport", "--source-port"):
+            return _match("sport", negated, _ports(value))
+        return None
+    if module == "multiport":
+        if option in ("--dports", "--destination-ports"):
+            return _match("dport", negated, _ports(value))
+        if option in ("--sports", "--source-ports"):
+            return _match("sport", negated, _ports(value))
+        return None
+    if module == "state" and option == "--state":
+        return _match("ct_state", negated, _states(value))
+    if module == "conntrack":
+        if option == "--ctstate":
+            return _match("ct_state", negated, _states(value))
+        if option == "--ctorigdstport":
+            return _match("ct_orig_dport", negated, _ports(value))
+    return None
+
+
+def _target(name: str, chains: Set[str]) -> Tuple[str, Optional[str]]:
+    if name in _TERMINAL:
+        return _TERMINAL[name], None
+    if name in chains:
+        return "jump", name
+    if name in _NON_TERMINAL:
+        return "continue", name
+    return "other", name
+
+
+def _iptables_rule(line: str, chains: Set[str]) -> Tuple[str, NetfilterRuleDict]:
+    """``-A CHAIN ...`` as ``(chain, rule)``."""
+    tokens = shlex.split(line)
+    chain = tokens[1]
+    rt = _RuleTokens(tokens[2:])
+    matches: List[NetfilterMatchDict] = []
+    verdict, target = "continue", None
+    negated = False
+    simple = {
+        "-i": "iifname",
+        "--in-interface": "iifname",
+        "-o": "oifname",
+        "--out-interface": "oifname",
+        "-s": "saddr",
+        "--source": "saddr",
+        "-d": "daddr",
+        "--destination": "daddr",
+        "-p": "l4proto",
+        "--protocol": "l4proto",
+    }
+    while rt.more():
+        token = rt.take()
+        if token == "!":
+            negated = True
+            continue
+        if token in simple:
+            field, value = simple[token], rt.take()
+            if field in ("iifname", "oifname"):
+                matches.append(_match(field, negated, [_iface(value)]))
+            elif field == "l4proto":
+                if value.lower() != "all":
+                    matches.append(_match(field, negated, [value.lower()]))
+            else:
+                matches.append(_match(field, negated, value.split(",")))
+        elif token in ("-m", "--match"):
+            _module_options(rt.take(), rt, matches)
+        elif token in ("-j", "--jump"):
+            verdict, target = _target(rt.take(), chains)
+            break  # the target's own options follow and match nothing
+        elif token in ("-g", "--goto"):
+            verdict, target = "goto", rt.take()
+            break
+        elif token in ("-c", "--set-counters"):
+            rt.take(), rt.take()
+        else:
+            matches.append(_unknown(token))
+        negated = False
+    return chain, {"matches": matches, "verdict": verdict, "target": target, "text": line}
+
+
+def _iptables_variant(lines: List[str]) -> Optional[str]:
+    """``iptables-nft`` or ``iptables-legacy`` from the header; None if nothing was saved."""
+    for line in lines:
+        if line.startswith("# Generated by"):
+            return "iptables-nft" if "(nf_tables)" in line else "iptables-legacy"
+    return None
+
+
+def _parse_iptables_save(lines: List[str], family: str) -> List[NetfilterChainDict]:
+    chains: List[NetfilterChainDict] = []
+    table: Optional[str] = None
+    by_name: Dict[str, NetfilterChainDict] = {}
+    for raw in lines:
+        line = raw.strip()
+        if line.startswith("*"):
+            table, by_name = line[1:], {}
+        elif line.startswith(":") and table is not None:
+            name, policy = line[1:].split()[:2]
+            kind, hook, prio = _BUILT_IN.get(table, {}).get(name, (None, None, None))
+            chain: NetfilterChainDict = {
+                "family": family,
+                "table": table,
+                "name": name,
+                "type": kind,
+                "hook": hook,
+                "priority": prio,
+                "policy": None if policy == "-" else policy.lower(),
+                "rules": [],
+            }
+            by_name[name] = chain
+            chains.append(chain)
+        elif line.startswith("-A ") and table is not None:
+            name, rule = _iptables_rule(line, set(by_name))
+            if name in by_name:
+                by_name[name]["rules"].append(rule)
+    return chains
+
+
+# ── nft -j ───────────────────────────────────────────────────────────────────
+
+_VERDICTS = {
+    "accept": "accept",
+    "drop": "drop",
+    "reject": "reject",
+    "return": "return",
+    "continue": "continue",
+}
+#: Statements that act on a packet without deciding its fate.
+_NO_OPS = frozenset(
+    {
+        "counter",
+        "log",
+        "mangle",
+        "notrack",
+        "set",
+        "flow",
+        "ct helper",
+        "synproxy",
+        "tcp option",
+        "dup",
+        "fwd",
+    }
+)
+#: Statements that pass only some packets on: what follows is not about every packet.
+_PARTIAL = frozenset({"limit", "quota", "numgen", "hash", "ct count"})
+#: Verdicts that hand the packet elsewhere, or that this does not model.
+_OTHER = frozenset({"queue", "dnat", "snat", "masquerade", "redirect", "tproxy", "vmap"})
+_TRANSPORTS = frozenset({"tcp", "udp", "sctp", "udplite", "dccp"})
+
+
+def _nft_scalar(value: Any) -> Optional[List[str]]:
+    """A right-hand side as values; None for one this does not model."""
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, (int, str)):
+        return [str(value)]
+    if isinstance(value, dict):
+        if "prefix" in value:
+            return [f"{value['prefix']['addr']}/{value['prefix']['len']}"]
+        if "range" in value:
+            low, high = value["range"]
+            return [f"{low}-{high}"]
+        if "elem" in value:
+            return _nft_scalar(value["elem"].get("val"))
+        if "set" in value:
+            return _nft_values(value["set"])
+    if isinstance(value, list):
+        return _nft_values(value)
+    return None
+
+
+def _nft_values(items: Any) -> Optional[List[str]]:
+    items = items if isinstance(items, list) else [items]
+    out: List[str] = []
+    for item in items:
+        values = _nft_scalar(item)
+        if values is None:
+            return None
+        out += values
+    return out
+
+
+def _left_text(left: Any) -> str:
+    if isinstance(left, dict):
+        if "payload" in left:
+            return (
+                f"{left['payload'].get('protocol', '')} {left['payload'].get('field', '')}".strip()
+            )
+        if "meta" in left:
+            return f"meta {left['meta'].get('key', '')}"
+        if "ct" in left:
+            ct = left["ct"]
+            return f"ct {ct.get('dir', '')} {ct.get('key', '')}".replace("  ", " ")
+        key = next(iter(left), "")
+        return str(key)
+    return str(left)
+
+
+def _right_text(right: Any) -> str:
+    values = _nft_scalar(right)
+    if values is None:
+        return json.dumps(right, sort_keys=True)
+    return values[0] if len(values) == 1 else "{" + ", ".join(values) + "}"
+
+
+def _nft_field(left: Any) -> Tuple[Optional[str], Optional[str]]:
+    """``(field, implied l4proto)`` of a left-hand side; ``(None, None)`` if not modelled."""
+    if not isinstance(left, dict):
+        return None, None
+    if "meta" in left:
+        key = left["meta"].get("key")
+        return {
+            "iifname": "iifname",
+            "iif": "iifname",
+            "oifname": "oifname",
+            "oif": "oifname",
+            "l4proto": "l4proto",
+            "nfproto": "nfproto",
+        }.get(key), None
+    if "payload" in left:
+        proto, field = left["payload"].get("protocol"), left["payload"].get("field")
+        if proto in ("ip", "ip6") and field in ("saddr", "daddr"):
+            return field, None
+        if (proto, field) in (("ip", "protocol"), ("ip6", "nexthdr")):
+            return "l4proto", None
+        if field in ("dport", "sport") and (proto in _TRANSPORTS or proto == "th"):
+            return field, proto if proto in _TRANSPORTS else None
+        return None, None
+    if "ct" in left:
+        key, direction = left["ct"].get("key"), left["ct"].get("dir")
+        if key == "state":
+            return "ct_state", None
+        if key == "proto-dst" and direction == "original":
+            return "ct_orig_dport", None
+    return None, None
+
+
+def _nft_match(
+    match: Dict[str, Any], sets: Dict[Tuple[str, str, str], List[str]], where: Tuple[str, str]
+) -> List[NetfilterMatchDict]:
+    left, op, right = match.get("left"), match.get("op"), match.get("right")
+    field, implied = _nft_field(left)
+    raw_text = (
+        f"{_left_text(left)} {'' if op in ('==', 'in') else str(op) + ' '}{_right_text(right)}"
+    )
+    if field is None or op not in ("==", "in", "!="):
+        return [_unknown(raw_text)]
+    if isinstance(right, str) and right.startswith("@"):
+        values = sets.get((where[0], where[1], right[1:]))
+    else:
+        values = _nft_scalar(right)
+    if values is None:
+        return [_unknown(raw_text)]
+    out: List[NetfilterMatchDict] = []
+    if implied:
+        out.append(_match("l4proto", False, [implied]))
+    if field in ("iifname", "oifname", "l4proto", "ct_state", "nfproto"):
+        values = [
+            str(v).lower() if field != "iifname" and field != "oifname" else str(v) for v in values
+        ]
+    out.append(_match(field, op == "!=", values))
+    return out
+
+
+def _nft_rule(
+    rule: Dict[str, Any], sets: Dict[Tuple[str, str, str], List[str]]
+) -> NetfilterRuleDict:
+    where = (rule.get("family", ""), rule.get("table", ""))
+    matches: List[NetfilterMatchDict] = []
+    verdict, target = "continue", None
+    for statement in rule.get("expr") or []:
+        if not isinstance(statement, dict) or not statement:
+            continue
+        key = next(iter(statement))
+        body = statement[key]
+        if key == "match":
+            matches += _nft_match(body, sets, where)
+        elif key in _VERDICTS:
+            verdict = _VERDICTS[key]
+            break
+        elif key in ("jump", "goto"):
+            verdict, target = key, str((body or {}).get("target"))
+            break
+        elif key == "xt":
+            if (body or {}).get("type") == "target":
+                verdict, target = "other", str(body.get("name"))
+                break
+            matches.append(_unknown(f"xt match {(body or {}).get('name')}"))
+        elif key in _PARTIAL:
+            matches.append(_unknown(key))
+        elif key in _OTHER:
+            verdict, target = "other", key
+            break
+        elif key not in _NO_OPS:
+            matches.append(_unknown(key))
+    entry: NetfilterRuleDict = {
+        "matches": matches,
+        "verdict": verdict,
+        "target": target,
+        "text": "",
+    }
+    entry["text"] = rule_text(entry)
+    return entry
+
+
+def _parse_nft(text: str, skip: Set[Tuple[str, str]]) -> List[NetfilterChainDict]:
+    start = text.find("{")
+    if start < 0:
+        raise ValueError("nft printed no JSON")
+    objects = json.loads(text[start:]).get("nftables") or []
+    sets: Dict[Tuple[str, str, str], List[str]] = {}
+    for obj in objects:
+        if "set" in obj and "elem" in obj["set"]:
+            values = _nft_values(obj["set"]["elem"])
+            if values is not None:
+                s = obj["set"]
+                sets[(s.get("family", ""), s.get("table", ""), s.get("name", ""))] = values
+    chains: Dict[Tuple[str, str, str], NetfilterChainDict] = {}
+    for obj in objects:
+        if "chain" not in obj:
+            continue
+        c = obj["chain"]
+        if (c.get("family"), c.get("table")) in skip:
+            continue
+        chains[(c["family"], c["table"], c["name"])] = {
+            "family": c["family"],
+            "table": c["table"],
+            "name": c["name"],
+            "type": c.get("type"),
+            "hook": c.get("hook"),
+            "priority": c.get("prio"),
+            "policy": c.get("policy") if c.get("hook") else None,
+            "rules": [],
+        }
+    for obj in objects:
+        if "rule" not in obj:
+            continue
+        r = obj["rule"]
+        chain = chains.get((r.get("family"), r.get("table"), r.get("chain")))
+        if chain is not None:
+            chain["rules"].append(_nft_rule(r, sets))
+    return list(chains.values())
+
+
+# ── text ─────────────────────────────────────────────────────────────────────
+
+
+def _family_of(values: List[str]) -> str:
+    try:
+        return (
+            "ip6"
+            if all(ipaddress.ip_network(v, strict=False).version == 6 for v in values)
+            else "ip"
+        )
+    except ValueError:
+        return "ip"
+
+
+def _values_text(values: List[str]) -> str:
+    return values[0] if len(values) == 1 else "{" + ", ".join(values) + "}"
+
+
+def rule_text(rule: NetfilterRuleDict) -> str:
+    """A rule as one nft-style line, from its normalized matches and verdict."""
+    protos = [
+        v
+        for m in rule["matches"]
+        if m["field"] == "l4proto" and m["op"] == "=="
+        for v in m["values"]
+    ]
+    parts: List[str] = []
+    for m in rule["matches"]:
+        neg = "!= " if m["op"] == "!=" else ""
+        field, values = m["field"], m["values"]
+        if field == "unknown":
+            parts.append(" ".join(values))
+            continue
+        keyword = {
+            "saddr": f"{_family_of(values)} saddr",
+            "daddr": f"{_family_of(values)} daddr",
+            "l4proto": "meta l4proto",
+            "nfproto": "meta nfproto",
+            "ct_state": "ct state",
+            "ct_orig_dport": "ct original proto-dst",
+            "dport": f"{protos[0] if len(protos) == 1 else 'th'} dport",
+            "sport": f"{protos[0] if len(protos) == 1 else 'th'} sport",
+        }.get(field, field)
+        parts.append(f"{keyword} {neg}{_values_text(values)}")
+    verdict = rule["verdict"]
+    if verdict in ("jump", "goto"):
+        parts.append(f"{verdict} {rule['target']}")
+    elif verdict == "other":
+        parts.append(str(rule["target"] or "other"))
+    elif verdict != "continue":
+        parts.append(verdict)
+    return " ".join(parts)
+
+
+# ── the reading ──────────────────────────────────────────────────────────────
+
+
+def _iptables_sections(
+    sections: Dict[str, Tuple[List[str], Optional[int]]],
+) -> Iterator[Tuple[str, str, List[str]]]:
+    """``(variant, family, lines)`` for each iptables tool that saved something."""
+    for tool, family in _IPTABLES_FAMILY.items():
+        lines, rc = sections.get(tool, ([], None))
+        if rc != 0:
+            continue
+        variant = _iptables_variant(lines)
+        if variant is not None:
+            yield variant, family, lines
+
+
+def parse_host_firewall(output: str) -> HostFirewallDict:
+    """The framed report of :data:`HOST_FIREWALL_COMMAND` as one ruleset.
+
+    :raises ValueError: if the report's frame is not intact.
+    :raises HostFirewallUnavailable: if neither tool could be read.
+    """
+    sections = _sections(_frame(output))
+    sources: List[str] = []
+    chains: List[NetfilterChainDict] = []
+    covered: Set[Tuple[str, str]] = set()
+    for variant, family, lines in _iptables_sections(sections):
+        saved = _parse_iptables_save(lines, family)
+        chains += saved
+        if variant not in sources:
+            sources.append(variant)
+        if variant == "iptables-nft":
+            covered |= {(c["family"], c["table"]) for c in saved}
+    nft_lines, nft_rc = sections.get("nft", ([], None))
+    if nft_rc == 0:
+        try:
+            nft_chains = _parse_nft("\n".join(nft_lines), covered)
+        except ValueError:
+            nft_chains = None
+        if nft_chains is not None:
+            sources.insert(0, "nftables")
+            chains = nft_chains + chains
+    if not sources:
+        raise HostFirewallUnavailable("neither nft nor iptables-save could be read (root needed)")
+    return {"sources": sources, "chains": chains}
+
+
+class HostFirewallMixin:
+    """Adds :meth:`get_host_firewall` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the command and its parse
+    are the same everywhere, so they are concrete here, and a driver supplies
+    only :meth:`_run_host_firewall_command` -- how a command reaches its host,
+    and how it gains root there. Mixed in by the drivers that can, so
+    ``hasattr(driver, "get_host_firewall")`` stays truthful.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_host_firewall_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root
+            when *privileged*. The command is a single ``sh -c`` invocation."""
+            ...
+
+    def get_host_firewall(self) -> HostFirewallDict:
+        """
+        Returns the host's netfilter rules: every chain with its hook, priority
+        and policy, and every rule as matches and a verdict.
+
+        * sources (list) - ``nftables``, ``iptables-nft``, ``iptables-legacy``: what was read
+        * chains (list) - see :class:`~napalm_device_types.models.NetfilterChainDict`
+
+        Example::
+
+            {
+                "sources": ["nftables", "iptables-nft"],
+                "chains": [
+                    {"family": "ip", "table": "filter", "name": "INPUT", "type": "filter",
+                     "hook": "input", "priority": 0, "policy": "accept",
+                     "rules": [
+                         {"matches": [{"field": "iifname", "op": "==", "values": ["ens6"]}],
+                          "verdict": "drop", "target": None,
+                          "text": "-A INPUT -i ens6 -j DROP"},
+                     ]},
+                ],
+            }
+
+        :raises HostFirewallUnavailable: if neither tool could be read -- no
+            ``nft``, no ``iptables-save``, or no root to run them.
+        :raises ValueError: if the report came back cut short.
+        """
+        command = f"sh -c {quote(HOST_FIREWALL_COMMAND)}"
+        return parse_host_firewall(self._run_host_firewall_command(command, privileged=True))

napalm_device_types/models.py

@@ -400,6 +400,62 @@ class ListeningSocketsDict(TypedDict):
     sockets: List[ListeningSocketDict]
 
 
+class NetfilterMatchDict(TypedDict):
+    """One condition of a netfilter rule.
+
+    ``field`` is one of ``iifname``, ``oifname``, ``saddr``, ``daddr``,
+    ``l4proto``, ``dport``, ``sport``, ``ct_state``, ``ct_orig_dport``,
+    ``nfproto`` -- or ``unknown``, whose single value is the condition's text.
+    The rule matches when the field's value is (``==``) or is not (``!=``) one
+    of ``values``: interface names (``eth*`` a wildcard), CIDRs, protocol names,
+    ports (``8000-8010`` a range), conntrack states in lower case.
+    """
+
+    field: str
+    op: str  # "==" or "!="
+    values: List[str]
+
+
+class NetfilterRuleDict(TypedDict):
+    """One rule: every match must hold, then the verdict applies.
+
+    ``verdict`` is ``accept``, ``drop``, ``reject``, ``return``, ``jump`` or
+    ``goto`` (to ``target``), ``continue`` (a rule that only counts, logs or
+    marks), or ``other`` -- NAT, a queue, a verdict map -- with ``target`` naming it.
+    """
+
+    matches: List[NetfilterMatchDict]
+    verdict: str
+    target: Optional[str]
+    text: str  # the rule as iptables-save printed it, or as one nft-style line
+
+
+class NetfilterChainDict(TypedDict):
+    """A chain. A base chain has a ``hook``, ``priority`` and ``policy``; a chain
+    reached only by ``jump``/``goto`` has None for all three."""
+
+    family: str  # "ip", "ip6", "inet", "bridge", "arp", "netdev"
+    table: str
+    name: str
+    type: Optional[str]  # "filter", "nat", "route"
+    hook: Optional[str]  # "prerouting", "input", "forward", "output", "postrouting", ...
+    priority: Optional[int]
+    policy: Optional[str]  # "accept" or "drop"
+    rules: List[NetfilterRuleDict]
+
+
+class HostFirewallDict(TypedDict):
+    """What ``HostFirewallMixin.get_host_firewall`` read.
+
+    ``sources`` names what was read: ``nftables``, ``iptables-nft`` (the tables
+    iptables manages, taken from ``iptables-save`` because nft shows their
+    extensions opaquely) and ``iptables-legacy`` (a ruleset of its own).
+    """
+
+    sources: List[str]
+    chains: List[NetfilterChainDict]
+
+
 class VirtualIpDict(TypedDict):
     """An address the host's HA configuration lets float between machines.