plugin linux_netfilter: fixes for port-ranges and unset mac-address

Author: superstes

src/firewall_test/plugins/system/firewall_netfilter.py

@@ -5,6 +5,8 @@
 from simulator.packet import PacketIP, PacketTCPUDP, PacketICMP
 from utils.logger import log_debug, log_warn
 
+# todo: add explicit match-tests
+
 
 # pylint: disable=R0912
 class RuleMatcherNetfilter(RuleMatcher):
@@ -16,95 +18,95 @@ def matches(self, packet: (PacketIP, PacketTCPUDP, PacketICMP), rule: Rule) -> R
         """
         nf_rule: NftRule = rule.raw
 
-        if rule.action is None:
+        if rule.action is None or \
+                not issubclass(rule.action, (RuleActionKindTerminal, RuleActionKindToChain, RuleActionKindNAT)):
             return RuleMatchResult(False, None, None, None, None)
 
-        if issubclass(rule.action, (RuleActionKindTerminal, RuleActionKindToChain, RuleActionKindNAT)):
-            results = []
-
-            if len(nf_rule.matches) == 0:
-                return RuleMatchResult(
-                    matched=True,
-                    action=rule.action,
-                    target_chain_name=nf_rule.target_chain,
-                    target_nat_ip=nf_rule.target_nat_ip,
-                    target_nat_port=nf_rule.target_nat_port,
-                )
-
-            for match in nf_rule.matches:
-                single_l3_result = True  # 'ip6 daddr != XXX' would drop any IPv4 traffic
-                single_results = []
-                # NETWORK INTERFACES
-                if match.match_ni_in:
-                    single_results.append(packet.ni_in in match.value)
-
-                if match.match_ni_out:
-                    single_results.append(packet.ni_out in match.value)
-
-                # IP PROTOCOL
-                if match.match_proto_l3:
-                    if match.value_proto_l3:
-                        single_l3_result = packet.proto_l3 == match.value_proto_l3
-
-                    else:
-                        single_l3_result = packet.proto_l3 in match.value
-
-                # IP SOURCE AND DESTINATION
-                if match.match_ip_saddr:
-                    single_results.append(any(
-                        packet.src in ip_net for ip_net in match.value
-                    ))
-
-                if match.match_ip_daddr:
-                    single_results.append(any(
-                        packet.dst in ip_net for ip_net in match.value
-                    ))
-
-                # TRANSPORT PROTOCOL
-                if match.match_proto_l4:
-                    if match.value_proto_l4:
-                        single_results.append(packet.proto_l4 == match.value_proto_l4)
-
-                    else:
-                        single_results.append(packet.proto_l4 in match.value)
-
-                if isinstance(packet, PacketTCPUDP):
-                    # PORTS
-                    if match.match_sport:
-                        single_results.append(packet.sport in match.value)
-
-                    if match.match_dport:
-                        single_results.append(packet.dport in match.value)
-
-                    # CONNECTION TRACKING STATE
-                    if match.match_ct:
-                        single_results.append(packet.ct in match.value)
-
-                # if we need to separate the L3-result from the actual condition as it can impact the match
-                results.append(single_l3_result)
-
-                if len(single_results) > 0:
-                    if match.operator in [match.OP_EQ, match.OP_IN]:
-                        results.append(all(single_results))
-
-                    elif match.operator in [match.OP_NE, match.OP_NOT]:
-                        results.append(not all(single_results))
-
-                    else:
-                        log_warn('Firewall Plugin', f' > Unable to get results for operator "{match.operator}"')
-
-            if len(results) == 0:
-                log_warn('Firewall Plugin', ' > Matches: Found not matches we could process - skipping rule')
-
-            else:
-                log_debug('Firewall Plugin', f' > Match Results: {nf_rule.get_match_types()} => {results}')
-
-                return RuleMatchResult(
-                    matched=all(results),
-                    action=rule.action,
-                    target_chain_name=nf_rule.target_chain,
-                    target_nat_ip=nf_rule.target_nat_ip,
-                    target_nat_port=nf_rule.target_nat_port,
-                )
+        results = []
+
+        if len(nf_rule.matches) == 0:
+            return RuleMatchResult(
+                matched=True,
+                action=rule.action,
+                target_chain_name=nf_rule.target_chain,
+                target_nat_ip=nf_rule.target_nat_ip,
+                target_nat_port=nf_rule.target_nat_port,
+            )
+
+        for match in nf_rule.matches:
+            single_l3_result = True  # 'ip6 daddr != XXX' would drop any IPv4 traffic
+            single_results = []
+            # NETWORK INTERFACES
+            if match.match_ni_in:
+                single_results.append(packet.ni_in in match.value)
+
+            if match.match_ni_out:
+                single_results.append(packet.ni_out in match.value)
+
+            # IP PROTOCOL
+            if match.match_proto_l3:
+                if match.value_proto_l3:
+                    single_l3_result = packet.proto_l3 == match.value_proto_l3
+
+                else:
+                    single_l3_result = packet.proto_l3 in match.value
+
+            # IP SOURCE AND DESTINATION
+            if match.match_ip_saddr:
+                single_results.append(any(
+                    packet.src in ip_net for ip_net in match.value
+                ))
+
+            if match.match_ip_daddr:
+                single_results.append(any(
+                    packet.dst in ip_net for ip_net in match.value
+                ))
+
+            # TRANSPORT PROTOCOL
+            if match.match_proto_l4:
+                if match.value_proto_l4:
+                    single_results.append(packet.proto_l4 == match.value_proto_l4)
+
+                else:
+                    single_results.append(packet.proto_l4 in match.value)
+
+            if isinstance(packet, PacketTCPUDP):
+                # PORTS
+                if match.match_sport:
+                    single_results.append(packet.sport in match.value)
+
+                if match.match_dport:
+                    single_results.append(packet.dport in match.value)
+
+                # CONNECTION TRACKING STATE
+                if match.match_ct:
+                    single_results.append(packet.ct in match.value)
+
+            # if we need to separate the L3-result from the actual condition as it can impact the match
+            results.append(single_l3_result)
+
+            if len(single_results) > 0:
+                if match.operator in [match.OP_EQ, match.OP_IN]:
+                    results.append(all(single_results))
+
+                elif match.operator in [match.OP_NE, match.OP_NOT]:
+                    results.append(not all(single_results))
+
+                else:
+                    log_warn('Firewall Plugin', f' > Unable to get results for operator "{match.operator}"')
+
+        if len(results) == 0:
+            log_warn('Firewall Plugin', ' > Matches: Found not matches we could process - skipping rule')
+
+        else:
+            log_debug('Firewall Plugin', f' > Match Results: {nf_rule.get_matches()} => {results}')
+
+            return RuleMatchResult(
+                matched=all(results),
+                action=rule.action,
+                target_chain_name=nf_rule.target_chain,
+                target_nat_ip=nf_rule.target_nat_ip,
+                target_nat_port=nf_rule.target_nat_port,
+            )
 
         return RuleMatchResult(False, None, None, None, None)

src/firewall_test/plugins/translate/linux.py

@@ -99,7 +99,7 @@ def get(self) -> list[NetworkInterface]:
     def _parse_ni(raw: dict) -> dict:
         r = {
             'name': raw['ifname'],
-            'mac': raw['address'],
+            'mac': raw.get('address', None),
             ProtoL3IP4.N: [],
             ProtoL3IP6.N: [],
             'net4': [],

src/firewall_test/plugins/translate/netfilter/elements.py

@@ -1,8 +1,8 @@
 from abc import ABC
-from ipaddress import ip_network, summarize_address_range, ip_address, IPv4Network, IPv6Network
+from ipaddress import ip_network, summarize_address_range, ip_address
 
 from config import ProtoL3, ProtoL3IP4, ProtoL3IP6, MatchPort, ProtoL4ICMP, ProtoL4TCP, ProtoL4UDP, ProtoL3IP4IP6, \
-    PROTO_L4_MAPPING, PROTO_L3_MAPPING, PROTOS_L3, PROTOS_L4
+    PROTO_L4_MAPPING, PROTO_L3_MAPPING
 from plugins.translate.netfilter.parts import RULE_ACTIONS, IGNORE_RULE_EXPRESSIONS, IGNORE_LEFT
 from utils.logger import log_warn
 
@@ -191,14 +191,21 @@ def parse_right(self):
                 return
 
             if 'prefix' in self._right:
+                # parsed in 'update_value_type'
+                self.value = [self._right]
+
+            if 'range' in self._right:
+                # parsed in 'update_value_type'
                 self.value = [self._right]
 
         if isinstance(self._right, (str, int)):
             self.value = [self._right]
 
         if self.value is None:
             if 'range' not in self._right and 'fin' not in self._right:
-                raise ValueError(self._right)
+                log_warn('Firewall Plugin', f'Unable to parse match (right): "{self._right}"')
+                return
+
             self.value = self._right
 
     def update_value_type(self):
@@ -220,11 +227,16 @@ def update_value_type(self):
                         values.append(ip_network(f"{v['addr']}/{v['len']}"))
 
                     elif 'range' in v:
-                        range_nets = list(summarize_address_range(
-                            ip_address(v['range'][0]),
-                            ip_address(v['range'][1]),
-                        ))
-                        values.extend(range_nets)
+                        if str(v['range'][0]).isnumeric():
+                            range_ports = range(v['range'][0], v['range'][1])
+                            values.extend(range_ports)
+
+                        else:
+                            range_nets = list(summarize_address_range(
+                                ip_address(v['range'][0]),
+                                ip_address(v['range'][1]),
+                            ))
+                            values.extend(range_nets)
 
                     else:
                         raise ValueError(self.value)
@@ -258,86 +270,54 @@ def update_value_type(self):
 
         self.value = values
 
-    def get_match_types(self) -> (list[str], None):
-        match = []
+    def get_matches(self) -> (dict, None):
+        matches = {}
         if self.match_proto_l3:
-            match.append('proto_l3')
+            if self.value_proto_l3:
+                matches['proto_l3'] = {self.OP_EQ: self.value_proto_l3.N}
+
+            else:
+                matches['proto_l3'] = {self.operator: [v.N for v in self.value]}
 
         if self.match_proto_l4:
-            match.append('proto_l4')
+            if self.value_proto_l4:
+                matches['proto_l4'] = {self.OP_EQ: self.value_proto_l4.N}
+
+            else:
+                matches['proto_l4'] = {self.operator: [v.N for v in self.value]}
 
         if self.match_ip_saddr:
-            match.append('ip_saddr')
+            matches['ip_saddr'] = { self.operator: [str(v) for v in self.value]}
 
         if self.match_ip_daddr:
-            match.append('ip_daddr')
+            matches['ip_daddr'] = {self.operator: [str(v) for v in self.value]}
 
         if self.match_ni_in:
-            match.append('ni_in')
+            matches['ni_in'] = {self.operator: self.value}
 
         if self.match_ni_out:
-            match.append('ni_out')
+            matches['ni_out'] = {self.operator: self.value}
 
         if self.match_sport:
-            match.append('src-port')
+            matches['src_port'] = {self.operator: self.value}
 
         if self.match_dport:
-            match.append('dst-port')
+            matches['dst_port'] = {self.operator: self.value}
 
         if self.match_ct:
-            match.append('ct')
+            matches['ct'] = {self.operator: self.value}
 
-        if len(match) == 0:
-            match = None
+        if len(matches) == 0:
+            return None
 
-        return match
+        return matches
 
     def __repr__(self) -> str:
-        values = []
-        for v in self.value:
-            if isinstance(v, (IPv4Network, IPv6Network)):
-                values.append(str(v))
-
-            elif v in PROTOS_L3 or v in PROTOS_L4:
-                values.append(v.N)
-
-            else:
-                values.append(v)
-
-        if self.value_proto_l3 is not None:
-            values.append(self.value_proto_l3.N)
-
-        if self.value_proto_l4 is not None:
-            values.append(self.value_proto_l4.N)
-
-        matches = self.get_match_types()
+        matches = self.get_matches()
         if matches is None:
             return 'None (unsupported)'
 
-        if len(matches) == 1:
-            return f"{matches[0]} {self.operator} {values}"
-
-        match_proto = None
-        value_proto = None
-        for proto_kind, proto_mapping in {
-            'proto_l3': PROTO_L3_MAPPING,
-            'proto_l4': PROTO_L4_MAPPING,
-        }.items():
-            if proto_kind in matches:
-                match_proto = proto_kind
-                for v in values:
-                    if v in proto_mapping:
-                        value_proto = v
-
-        if match_proto is not None and value_proto is not None:
-            matches.remove(match_proto)
-            values.remove(value_proto)
-            if len(matches) == 1:
-                matches = matches[0]
-
-            return f"{match_proto} {self.OP_EQ} {value_proto} & {matches} {self.operator} {values}"
-
-        return f"{matches} {self.operator} {values}"
+        return f"{self.operator} {matches}"
 
 
 class NftRule(NftBase):
@@ -349,6 +329,7 @@ def __init__(self, table: NftTable, chain: NftChain, raw: dict, seq: int, sets:
 
         self.comment = raw.get('comment', None)
         self.family = raw.get('family', None)
+        self.handle = raw.get('handle', None)
 
         self.action = None
         self.matches: list[NftMatch] = []
@@ -411,7 +392,7 @@ def _init_match(self, e: dict) -> bool:
                 left=e['match']['left'],
                 right=e['match']['right'],
             )
-            if match.get_match_types() is None:
+            if match.get_matches() is None:
                 self.invalid_matches = True
 
             else:
@@ -448,10 +429,10 @@ def _init_nat_xt(self, e: dict) -> bool:
 
         return False
 
-    def get_match_types(self) -> list[str]:
-        matches = []
+    def get_matches(self) -> dict:
+        matches = {}
         for m in self.matches:
-            matches.extend(m.get_match_types())
+            matches = {**matches, **m.get_matches()}
 
         return matches