Merge pull request 'feat: HostFirewallMixin, a Linux host's own netfilter rules (NetOrk/netork#1344)' (#51) from feat/host-firewall-1344 into main
This commit was merged in pull request #51.
This commit is contained in:
christianmanivong committed
commit
ac8e7c4c11
6 files changed
+1342
-1
No files matched your search
@@ -150,6 +150,17 @@ without `ss` is read with `netstat -lntup` (OpenWrt's busybox, old net-tools); o
|
||||
cgroup names the procd service (`/services/<name>/<instance>`). 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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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))
|
||||
@@ -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.
|
||||
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user