From f77091c489d4ef676237affe1ea84b3b52037043 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 21 Oct 2026 23:10:44 +0200 Subject: [PATCH] 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. --- README.md | 11 + napalm_device_types/__init__.py | 9 + napalm_device_types/host_firewall.py | 764 +++++++++++++++++++++++++++ napalm_device_types/models.py | 56 ++ pyproject.toml | 2 +- tests/test_host_firewall.py | 501 ++++++++++++++++++ 6 files changed, 1342 insertions(+), 1 deletion(-) create mode 100644 napalm_device_types/host_firewall.py create mode 100644 tests/test_host_firewall.py diff --git a/README.md b/README.md index 0b2ba42..e7c6e8f 100644 --- a/README.md +++ b/README.md @@ -150,6 +150,17 @@ without `ss` is read with `netstat -lntup` (OpenWrt's busybox, old net-tools); o cgroup names the procd service (`/services//`). A host with neither raises `ListeningSocketsUnavailable`. +`HostFirewallMixin` (`get_host_firewall`) reads a Linux host's own netfilter rules the same +way: `nft -j list ruleset`, with the tables iptables-nft manages taken from `iptables-save` +instead, because nft shows their extensions (`-m conntrack`, `-m set`) only as an opaque `xt`. +A legacy iptables is a ruleset of its own and is kept beside nft's. Every chain says its family, +table, hook, priority and policy; every rule is a list of matches (`iifname`, `saddr`, `dport`, +`ct_state`, ...) and a verdict. A condition it does not model is an `unknown` match with its +text, never left out. A driver supplies `_run_host_firewall_command(command, privileged=)`; both +tools need root, so a host where neither could be read raises `HostFirewallUnavailable` rather +than reading as one without rules. Whether the rules keep a service unreachable is the +consumer's call. + `VirtualIpsMixin` (`get_virtual_ips`) reads the addresses a host's HA configuration lets float between machines: vip-manager (5.x's `--config` YAML, 1.x's `-ip=${VIP_IP}` resolved through the unit's environment) and keepalived (`virtual_ipaddress` of every diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index e9a3bf1..3c8028a 100644 --- a/napalm_device_types/__init__.py +++ b/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", diff --git a/napalm_device_types/host_firewall.py b/napalm_device_types/host_firewall.py new file mode 100644 index 0000000..06211e4 --- /dev/null +++ b/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)) diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index d72634f..9ec1641 100644 --- a/napalm_device_types/models.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 7bb03aa..8b0471a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "4.20.0" +version = "4.21.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_host_firewall.py b/tests/test_host_firewall.py new file mode 100644 index 0000000..2bcd805 --- /dev/null +++ b/tests/test_host_firewall.py @@ -0,0 +1,501 @@ +"""get_host_firewall: a Linux host's own netfilter rules, read once, in one shape. + +Whether a service on a host is reachable from the internet is not only decided +by the address it listens on: the host's own firewall may drop the packets +before they reach it. Reading that is the same on every Linux host, so the +command and the parse live here, and a driver only carries the command across +(NetOrk/netork#1344). + +``nft -j list ruleset`` is the primary reading. Rules that iptables-nft wrote +show their extensions there only as an opaque ``xt`` (``-m conntrack``, +``-m set``), so the tables iptables manages are taken from ``iptables-save``, +which spells them out. A legacy iptables is a ruleset of its own beside nft, +and both apply. +""" + +from __future__ import annotations + +import json +from shlex import quote + +import pytest + +from napalm_device_types.host_firewall import ( + HOST_FIREWALL_COMMAND, + HostFirewallMixin, + HostFirewallUnavailable, + parse_host_firewall, +) + +IPTABLES_NFT = """# Generated by iptables-save v1.8.10 (nf_tables) on Wed Oct 21 14:52:11 2026 +*filter +:INPUT ACCEPT [2783578:295175610] +:FORWARD ACCEPT [0:0] +:OUTPUT ACCEPT [0:0] +:DOCKER-USER - [0:0] +:WGRUNNER - [0:0] +-A INPUT -s 10.255.255.4/32 -j WGRUNNER +-A INPUT -i ens6 -p udp -m udp --dport 51820 -j ACCEPT +-A INPUT -m set --match-set crowdsec-blacklists src -j DROP +-A INPUT -i ens6 -m state --state RELATED,ESTABLISHED -j ACCEPT +-A INPUT -i ens6 -p tcp -m multiport --dports 80,443,8000:8010 -j ACCEPT +-A INPUT ! -i lo -p tcp -m conntrack --ctstate NEW --ctorigdstport 2222 -j LOG --log-prefix "ssh " +-A INPUT -i ens6 -j DROP +-A FORWARD -j DOCKER-USER +-A DOCKER-USER -i eth+ ! -s 192.168.0.0/16 -g WGRUNNER +-A WGRUNNER -d 10.7.224.12/32 -p tcp -m tcp --dport 10051 -j ACCEPT +-A WGRUNNER -j REJECT --reject-with icmp-port-unreachable +-A WGRUNNER -j RETURN +COMMIT +# Completed on Wed Oct 21 14:52:11 2026 +*nat +:PREROUTING ACCEPT [0:0] +:POSTROUTING ACCEPT [0:0] +:DOCKER - [0:0] +-A PREROUTING -m addrtype --dst-type LOCAL -j DOCKER +-A DOCKER ! -i br-1 -p udp -m udp --dport 53 -j DNAT --to-destination 172.18.0.2:53 +COMMIT +""" + +IP6TABLES_NFT = """# Generated by ip6tables-save v1.8.10 (nf_tables) on Wed Oct 21 14:52:11 2026 +*filter +:INPUT DROP [0:0] +:FORWARD ACCEPT [0:0] +:OUTPUT ACCEPT [0:0] +-A INPUT -i lo -j ACCEPT +COMMIT +""" + + +def _nft(*objects: dict) -> str: + meta = {"metainfo": {"version": "1.0.9", "json_schema_version": 1}} + return json.dumps({"nftables": [meta, *objects]}) + + +#: The iptables-nft filter table as nft shows it -- with its extensions opaque -- +#: and a native ``inet filter`` table of the host's own nftables.service. +NFT = _nft( + {"table": {"family": "ip", "name": "filter", "handle": 1}}, + { + "chain": { + "family": "ip", + "table": "filter", + "name": "INPUT", + "type": "filter", + "hook": "input", + "prio": 0, + "policy": "accept", + } + }, + { + "rule": { + "family": "ip", + "table": "filter", + "chain": "INPUT", + "expr": [ + {"xt": {"type": "match", "name": "set"}}, + {"counter": {"packets": 0, "bytes": 0}}, + {"drop": None}, + ], + } + }, + {"table": {"family": "inet", "name": "host", "handle": 9}}, + { + "set": { + "family": "inet", + "table": "host", + "name": "admins", + "type": "ipv4_addr", + "flags": ["interval"], + "elem": [{"prefix": {"addr": "198.51.100.0", "len": 24}}], + } + }, + { + "chain": { + "family": "inet", + "table": "host", + "name": "input", + "type": "filter", + "hook": "input", + "prio": 10, + "policy": "drop", + } + }, + {"chain": {"family": "inet", "table": "host", "name": "services"}}, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + { + "match": { + "op": "in", + "left": {"ct": {"key": "state"}}, + "right": ["established", "related"], + } + }, + {"accept": None}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + {"match": {"op": "==", "left": {"meta": {"key": "iif"}}, "right": "lo"}}, + {"accept": None}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + { + "match": { + "op": "==", + "left": {"payload": {"protocol": "ip", "field": "saddr"}}, + "right": "@admins", + } + }, + { + "match": { + "op": "==", + "left": {"payload": {"protocol": "tcp", "field": "dport"}}, + "right": {"set": [22, {"range": [8000, 8010]}]}, + } + }, + {"accept": None}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + {"match": {"op": "!=", "left": {"meta": {"key": "iifname"}}, "right": "wg*"}}, + { + "match": { + "op": "==", + "left": {"payload": {"protocol": "ip6", "field": "daddr"}}, + "right": {"prefix": {"addr": "2001:db8::", "len": 32}}, + } + }, + {"jump": {"target": "services"}}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + {"match": {"op": "==", "left": {"meta": {"key": "l4proto"}}, "right": "udp"}}, + {"limit": {"rate": 10, "per": "second"}}, + {"log": {"prefix": "udp "}}, + {"reject": {"type": "icmpx", "expr": "port-unreachable"}}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "input", + "expr": [ + { + "match": { + "op": "==", + "left": {"fib": {"result": "type", "flags": ["daddr"]}}, + "right": "local", + } + }, + {"goto": {"target": "services"}}, + ], + } + }, + { + "rule": { + "family": "inet", + "table": "host", + "chain": "services", + "expr": [ + { + "match": { + "op": "==", + "left": {"payload": {"protocol": "udp", "field": "dport"}}, + "right": 53, + } + }, + {"queue": {"num": 1}}, + ], + } + }, +) + + +def _report( + *, + nft: str | None = NFT, + nft_rc: int = 0, + ipt: str | None = IPTABLES_NFT, + ip6t: str | None = IP6TABLES_NFT, +) -> str: + parts = ["$ sh -c ...", "FW_BEGIN"] + if nft is not None: + parts += ["[nft]", nft, f"__FW_RC={nft_rc}"] + if ipt is not None: + parts += ["[iptables-save]", ipt, "__FW_RC=0"] + if ip6t is not None: + parts += ["[ip6tables-save]", ip6t, "__FW_RC=0"] + parts.append("FW_END") + return "\n".join(parts) + "\n" + + +def _chain(reading, family, table, name): + (chain,) = [ + c + for c in reading["chains"] + if (c["family"], c["table"], c["name"]) == (family, table, name) + ] + return chain + + +def _fields(rule): + return [(m["field"], m["op"], m["values"]) for m in rule["matches"]] + + +class TestTheSources: + def test_iptables_tables_come_from_iptables_save_the_rest_from_nft(self): + reading = parse_host_firewall(_report()) + + assert reading["sources"] == ["nftables", "iptables-nft"] + tables = {(c["family"], c["table"]) for c in reading["chains"]} + assert tables == {("ip", "filter"), ("ip", "nat"), ("ip6", "filter"), ("inet", "host")} + # The filter table's INPUT is iptables-save's: its set match is spelled + # out there, and nft's opaque copy is not added a second time. + assert len(_chain(reading, "ip", "filter", "INPUT")["rules"]) == 7 + + def test_a_legacy_iptables_is_a_ruleset_of_its_own(self): + legacy = IPTABLES_NFT.replace("(nf_tables)", "(legacy)") + + reading = parse_host_firewall(_report(ipt=legacy, ip6t=None)) + + assert reading["sources"] == ["nftables", "iptables-legacy"] + inputs = [c for c in reading["chains"] if c["name"] == "INPUT" and c["family"] == "ip"] + assert len(inputs) == 2 # nft's own ip filter INPUT and legacy's both apply + + def test_nft_alone(self): + reading = parse_host_firewall(_report(ipt=None, ip6t=None)) + + assert reading["sources"] == ["nftables"] + assert _chain(reading, "ip", "filter", "INPUT")["rules"][0]["matches"] == [ + {"field": "unknown", "op": "==", "values": ["xt match set"]} + ] + + def test_iptables_alone_when_nft_fails(self): + reading = parse_host_firewall(_report(nft="Operation not permitted", nft_rc=1)) + + assert reading["sources"] == ["iptables-nft"] + + def test_no_tool_that_could_be_read(self): + with pytest.raises(HostFirewallUnavailable): + parse_host_firewall(_report(nft=None, ipt=None, ip6t=None)) + + def test_a_failed_reading_is_no_reading(self): + with pytest.raises(HostFirewallUnavailable): + parse_host_firewall(_report(nft="permission denied", nft_rc=1, ipt=None, ip6t=None)) + + def test_a_report_cut_short_raises(self): + with pytest.raises(ValueError): + parse_host_firewall(_report().replace("FW_END", "")) + + def test_an_empty_ruleset_is_a_host_without_rules(self): + reading = parse_host_firewall(_report(nft=_nft(), ipt=None, ip6t=None)) + + assert reading == {"sources": ["nftables"], "chains": []} + + +class TestIptablesSave: + def setup_method(self): + self.reading = parse_host_firewall(_report(nft=None)) + + def test_a_built_in_chain_has_its_hook_and_policy(self): + chain = _chain(self.reading, "ip", "filter", "INPUT") + assert (chain["type"], chain["hook"], chain["priority"], chain["policy"]) == ( + "filter", + "input", + 0, + "accept", + ) + assert _chain(self.reading, "ip6", "filter", "INPUT")["policy"] == "drop" + nat = _chain(self.reading, "ip", "nat", "PREROUTING") + assert (nat["type"], nat["hook"], nat["priority"]) == ("nat", "prerouting", -100) + + def test_a_user_chain_has_none(self): + chain = _chain(self.reading, "ip", "filter", "WGRUNNER") + assert (chain["type"], chain["hook"], chain["priority"], chain["policy"]) == ( + None, + None, + None, + None, + ) + + def test_matches_and_verdicts(self): + rules = _chain(self.reading, "ip", "filter", "INPUT")["rules"] + assert _fields(rules[0]) == [("saddr", "==", ["10.255.255.4/32"])] + assert (rules[0]["verdict"], rules[0]["target"]) == ("jump", "WGRUNNER") + assert _fields(rules[1]) == [ + ("iifname", "==", ["ens6"]), + ("l4proto", "==", ["udp"]), + ("dport", "==", ["51820"]), + ] + assert rules[1]["verdict"] == "accept" + assert _fields(rules[2]) == [ + ("unknown", "==", ["-m set --match-set crowdsec-blacklists src"]), + ] + assert _fields(rules[3]) == [ + ("iifname", "==", ["ens6"]), + ("ct_state", "==", ["related", "established"]), + ] + assert _fields(rules[4]) == [ + ("iifname", "==", ["ens6"]), + ("l4proto", "==", ["tcp"]), + ("dport", "==", ["80", "443", "8000-8010"]), + ] + assert _fields(rules[5]) == [ + ("iifname", "!=", ["lo"]), + ("l4proto", "==", ["tcp"]), + ("ct_state", "==", ["new"]), + ("ct_orig_dport", "==", ["2222"]), + ] + assert rules[5]["verdict"] == "continue" # LOG decides nothing + assert rules[6]["verdict"] == "drop" + + def test_negation_wildcard_and_goto(self): + (rule,) = _chain(self.reading, "ip", "filter", "DOCKER-USER")["rules"] + assert _fields(rule) == [("iifname", "==", ["eth*"]), ("saddr", "!=", ["192.168.0.0/16"])] + assert (rule["verdict"], rule["target"]) == ("goto", "WGRUNNER") + + def test_reject_return_and_nat_targets(self): + rules = _chain(self.reading, "ip", "filter", "WGRUNNER")["rules"] + assert [r["verdict"] for r in rules] == ["accept", "reject", "return"] + (dnat,) = _chain(self.reading, "ip", "nat", "DOCKER")["rules"] + assert (dnat["verdict"], dnat["target"]) == ("other", "DNAT") + assert _chain(self.reading, "ip", "nat", "PREROUTING")["rules"][0]["matches"] == [ + {"field": "unknown", "op": "==", "values": ["-m addrtype --dst-type LOCAL"]} + ] + + def test_the_line_is_kept_as_the_rules_text(self): + rule = _chain(self.reading, "ip", "filter", "INPUT")["rules"][6] + assert rule["text"] == "-A INPUT -i ens6 -j DROP" + + +class TestNftJson: + def setup_method(self): + self.chains = { + c["name"]: c + for c in parse_host_firewall(_report(ipt=None, ip6t=None))["chains"] + if c["table"] == "host" + } + + def test_a_base_chain_and_a_regular_one(self): + base, regular = self.chains["input"], self.chains["services"] + assert (base["family"], base["type"], base["hook"], base["priority"], base["policy"]) == ( + "inet", + "filter", + "input", + 10, + "drop", + ) + assert (regular["hook"], regular["policy"]) == (None, None) + + def test_matches(self): + rules = self.chains["input"]["rules"] + assert _fields(rules[0]) == [("ct_state", "==", ["established", "related"])] + assert _fields(rules[1]) == [("iifname", "==", ["lo"])] + # A named set is resolved from its elements; tcp dport implies tcp. + assert _fields(rules[2]) == [ + ("saddr", "==", ["198.51.100.0/24"]), + ("l4proto", "==", ["tcp"]), + ("dport", "==", ["22", "8000-8010"]), + ] + assert _fields(rules[3]) == [("iifname", "!=", ["wg*"]), ("daddr", "==", ["2001:db8::/32"])] + assert (rules[3]["verdict"], rules[3]["target"]) == ("jump", "services") + + def test_a_statement_that_takes_only_some_packets_is_unknown(self): + rule = self.chains["input"]["rules"][4] + assert _fields(rule) == [("l4proto", "==", ["udp"]), ("unknown", "==", ["limit"])] + assert rule["verdict"] == "reject" + + def test_an_expression_it_does_not_model_is_unknown(self): + rule = self.chains["input"]["rules"][5] + assert rule["matches"][0]["field"] == "unknown" + assert (rule["verdict"], rule["target"]) == ("goto", "services") + + def test_a_verdict_it_does_not_model_is_other(self): + (rule,) = self.chains["services"]["rules"] + assert rule["verdict"] == "other" + + def test_each_rule_reads_as_a_line(self): + assert self.chains["input"]["rules"][2]["text"] == ( + "ip saddr 198.51.100.0/24 meta l4proto tcp tcp dport {22, 8000-8010} accept" + ) + + def test_an_unresolvable_set_is_unknown(self): + nft = NFT.replace('"@admins"', '"@missing"') + rules = [ + c + for c in parse_host_firewall(_report(nft=nft, ipt=None, ip6t=None))["chains"] + if c["name"] == "input" + ][0]["rules"] + assert rules[2]["matches"][0] == { + "field": "unknown", + "op": "==", + "values": ["ip saddr @missing"], + } + + +class _Host(HostFirewallMixin): + def __init__(self, output: str | None = None, fail: Exception | None = None): + self.calls: list[tuple[str, bool]] = [] + self._output, self._fail = output, fail + + def _run_host_firewall_command(self, command: str, *, privileged: bool) -> str: + self.calls.append((command, privileged)) + if self._fail: + raise self._fail + return self._output or "" + + +class TestTheMixin: + def test_one_privileged_command(self): + host = _Host(_report()) + + reading = host.get_host_firewall() + + assert host.calls == [(f"sh -c {quote(HOST_FIREWALL_COMMAND)}", True)] + assert reading["sources"] == ["nftables", "iptables-nft"] + + def test_without_root_there_is_nothing_to_read(self): + # nft and iptables-save both need CAP_NET_ADMIN; a reading without root + # would be an empty ruleset that is not the host's. + host = _Host("FW_BEGIN\n[nft]\nnetlink: Operation not permitted\n__FW_RC=1\nFW_END\n") + + with pytest.raises(HostFirewallUnavailable): + host.get_host_firewall() + assert len(host.calls) == 1 + + def test_the_command_is_framed_and_read_only(self): + assert "nft -j list ruleset" in HOST_FIREWALL_COMMAND + assert "iptables-save" in HOST_FIREWALL_COMMAND + for word in (" -F", " -X", "flush", "delete", "-A ", "-I ", "restore"): + assert word not in HOST_FIREWALL_COMMAND -- 2.54.0