plugin opnsense: add basic rule-matching logic

Author: superstes

docs/source/plugins/firewall_netfilter.rst

@@ -99,3 +99,4 @@ These rule-expressions are unsupported for now:
 * :code:`ct helper`
 * :code:`&`
 * TCP flags
+* ICMP Types (codes are supported)

docs/source/plugins/firewall_opnsense.rst

@@ -38,7 +38,7 @@ Here is an example on how to run it with the exported config:
             --file-ruleset 'config.xml' \
             --file-interfaces 'network.json' \
             --file-routes 'network.json' \
-            --src-ip 172.17.11.5 \
+            --src-ip 10.57.65.44 \
             --dst-ip 1.1.1.1
 
 ----
@@ -52,3 +52,46 @@ Source Code
 
 * **Traffic Matching**: `system/firewall_opnsense.py <https://github.com/O-X-L/firewall-testing-framework/blob/latest/src/firewall_test/plugins/system/firewall_opnsense.py>`_
 
+----
+
+Config-Parsing
+##############
+
+The current implementation focused on supporting the most widely used rule matches like:
+
+* Layer 3 Protocol (IPv4/IPv6)
+* Layer 4 Protocol (tcp/udp/icmp)
+* Source- and Destination-IP filters
+* Source- and Destination-Port filters
+* Source- and Destination-NAT (including masquerade)
+* Inbound- and Outbound-Network-Interfaces
+* Aliases (including IPs/networks, ports, port-ranges, nested aliases, interface-groups, interface-IPs and -networks, :code:`This Firewall`, bogons, DNS-resolved hosts, URLTable/IPLists, ...)
+* Quick/Lazy actions
+
+The main match-parsing logic can be found here: `translate/opnsense/ruleset.py <https://github.com/O-X-L/firewall-testing-framework/tree/latest/src/firewall_test/plugins/translate/opnsense/ruleset.py>`_ & `translate/opnsense/rule.py <https://github.com/O-X-L/firewall-testing-framework/tree/latest/src/firewall_test/plugins/translate/opnsense/rule.py>`_
+
+If we were not able to parse any match from the rule-config - the rule will be skipped.
+
+If this happens you will see a warning at runtime! (:code:`Unsupported rule`)
+
+.. warning::
+
+    The plugin does not validate the syntax of the config-export you provide!
+
+    You may encounter unexpected errors when manually modifying it!
+
+Unsupported Expressions
+=======================
+
+Rules containing unsupported expressions will be skipped for now.
+
+If this happens you will see a warning at runtime! (:code:`Unsupported rule-expression`)
+
+These rule-expressions are unsupported for now:
+
+* :code:`GeoIP aliases`
+* :code:`URLTable JSON`
+* :code:`ICMP Types`
+* :code:`tagging`
+* TCP flags
+* Connection states

src/firewall_test/config.py

@@ -74,7 +74,7 @@ class ProtoL3IP4IP6(ProtoL3):
     N = 'ip'
 
 
-PROTOS_L3 = [ProtoL3IP4, ProtoL3IP6, ProtoL3IP4IP6]
+PROTOS_L3 = (ProtoL3IP4, ProtoL3IP6, ProtoL3IP4IP6)
 PROTO_L3_MAPPING = {
     ProtoL3IP4.N: ProtoL3IP4,
     ProtoL3IP6.N: ProtoL3IP6,
@@ -97,7 +97,7 @@ class ProtoL4ICMP(ProtoL4):
     N = 'icmp'
 
 
-PROTOS_L4 = [ProtoL4TCP, ProtoL4UDP, ProtoL4ICMP]
+PROTOS_L4 = (ProtoL4TCP, ProtoL4UDP, ProtoL4ICMP)
 PROTO_L4_MAPPING = {
     ProtoL4TCP.N: ProtoL4TCP,
     ProtoL4UDP.N: ProtoL4UDP,
@@ -190,11 +190,11 @@ class RuleActionSNAT(RuleActionKindNAT):
     N = 'snat'
 
 
-RULE_ACTIONS = [
+RULE_ACTIONS = (
     RuleActionAccept, RuleActionDrop, RuleActionReject,
     RuleActionJump, RuleActionGoTo, RuleActionContinue, RuleActionReturn,
     RuleActionDNAT, RuleActionSNAT,
-]
+)
 RULE_ACTION_MAPPING = {
     RuleActionAccept.N: RuleActionAccept,
     RuleActionDrop.N: RuleActionDrop,

src/firewall_test/plugins/system/firewall_opnsense.py

@@ -1,16 +1,102 @@
-from simulator.packet import PacketIP
+from config import ProtoL3IP4IP6
+from simulator.packet import PACKET_KINDS, PacketTCPUDP
 from plugins.translate.abstract import Rule
 from plugins.system.abstract_rule_match import RuleMatcher, RuleMatchResult
 from plugins.translate.opnsense.rule import OPNsenseRule
+from utils.logger import log_warn, log_debug
 
+# todo: add explicit match-tests
 
+
+# pylint: disable=R0912
 class RuleMatcherOPNsense(RuleMatcher):
-    def matches(self, packet: PacketIP, rule: Rule) -> RuleMatchResult:
+    def matches(self, packet: PACKET_KINDS, rule: Rule) -> RuleMatchResult:
         """
         :param packet: Packet to match
         :param rule: Rule to check
         :return: RuleMatchResult
         """
         opn_rule: OPNsenseRule = rule.raw
-        del opn_rule
+
+        results = []
+        ### NETWORK INTERFACES ###
+        if opn_rule.match_ni_in:
+            results.append(packet.ni_in in opn_rule.nis)
+
+        if opn_rule.match_ni_out:
+            results.append(packet.ni_out in opn_rule.nis)
+
+        ### PROTOCOLS ###
+        if opn_rule.ipp is not None:
+            if opn_rule.ipp == ProtoL3IP4IP6:
+                results.append(True)
+
+            else:
+                results.append(opn_rule.ipp == packet.proto_l3)
+
+        if opn_rule.proto is not None:
+            if not hasattr(packet, 'proto_l4'):
+                results.append(False)
+
+            else:
+                results.append(packet.proto_l4 in opn_rule.proto)
+
+        ### SOURCE / DESTINATION ###
+        if opn_rule.match_ip_saddr:
+            if opn_rule.src_any:
+                match = True
+
+            else:
+                match = any(packet.src in net for net in opn_rule.src)
+
+            if opn_rule.src_invert:
+                results.append(not match)
+
+            else:
+                results.append(match)
+
+        if opn_rule.match_ip_daddr:
+            if opn_rule.dst_any:
+                match = True
+
+            else:
+                match = any(packet.dst in net for net in opn_rule.dst)
+
+            if opn_rule.dst_invert:
+                results.append(not match)
+
+            else:
+                results.append(match)
+
+        ### PORTS ###
+        if opn_rule.src_port is not None and len(opn_rule.src_port) > 0:
+            if not isinstance(packet, PacketTCPUDP):
+                results.append(False)
+
+            else:
+                results.append(packet.sport in opn_rule.src_port)
+
+        if opn_rule.dst_port is not None and len(opn_rule.dst_port) > 0:
+            if not isinstance(packet, PacketTCPUDP):
+                results.append(False)
+
+            else:
+                results.append(packet.dport in opn_rule.dst_port)
+
+        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: {opn_rule.get_matches()} => {results}')
+
+            return RuleMatchResult(
+                matched=all(results),
+                action=rule.action,
+                target_chain_name=None,
+                target_nat_ip=None,
+                target_nat_port=None,
+            )
+
+        # todo: SNAT / DNAT
+
         return RuleMatchResult(False, None, None, None, None)

src/firewall_test/plugins/translate/abstract.py

@@ -266,6 +266,9 @@ def dump(self) -> dict:
             'raw': self.raw,
         }
 
+    def log(self) -> str:
+        return f'Seq {self.seq}, Action: {self.action.N}, {self.raw}'
+
     def validate(self):
         r = self.dump()
         if self.action is not None: