added basic opnsense rule-parsing

superstes · superstes · commit 3daabbf5314a · 2025-09-08T23:35:49.000+02:00
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
@@ -9,12 +9,14 @@ on:
     paths:
       - '**.py'
       - '.github/workflows/lint.yml'
+      - 'requirements.txt'
       - 'requirements_lint.txt'
   pull_request:
     branches: [latest]
     paths:
       - '**.py'
       - '.github/workflows/lint.yml'
+      - 'requirements.txt'
       - 'requirements_lint.txt'
 jobs:
   lint:
diff --git a/docs/source/plugins/firewall_opnsense.rst b/docs/source/plugins/firewall_opnsense.rst
@@ -1,10 +1,10 @@
 .. _plugins_fw_opnsense:
 
 .. |export_backup| image:: ../_static/img/plugin-opnsense-backup.png
-   :class: wiki-img
+   :class: wiki-img-sm
 
 .. |export_network| image:: ../_static/img/plugin-opnsense-export.png
-   :class: wiki-img
+   :class: wiki-img-sm
 
 .. include:: ../_include/head.rst
 
diff --git a/requirements.txt b/requirements.txt
@@ -0,0 +1,2 @@
+oxl-utils
+requests
diff --git a/src/firewall_test/config.py b/src/firewall_test/config.py
@@ -11,6 +11,47 @@
 DEFAULT_ROUTE_IP6 = ip_network('::/0')
 DEFAULT_ROUTES = [DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6]
 LINK_LOCAL_IP6 = ip_network('fe80::/64')
+BOGONS_IP4 = [
+    ip_network('0.0.0.0/8'),
+    ip_network('10.0.0.0/8'),
+    ip_network('100.64.0.0/10'),
+    ip_network('127.0.0.0/8'),
+    ip_network('127.0.53.53'),
+    ip_network('169.254.0.0/16'),
+    ip_network('172.16.0.0/12'),
+    ip_network('192.0.0.0/24'),
+    ip_network('192.0.2.0/24'),
+    ip_network('192.168.0.0/16'),
+    ip_network('198.18.0.0/15'),
+    ip_network('198.51.100.0/24'),
+    ip_network('203.0.113.0/24'),
+    ip_network('224.0.0.0/4'),
+    ip_network('240.0.0.0/4'),
+    ip_network('255.255.255.255/32'),
+]
+BOGONS_IP6 = [
+    ip_network('::/128'),
+    ip_network('::1/128'),
+    ip_network('::ffff:0:0/96'),
+    ip_network('::/96'),
+    ip_network('100::/64'),
+    ip_network('2001:10::/28'),
+    ip_network('2001:db8::/32'),
+    ip_network('3fff::/20'),
+    ip_network('fc00::/7'),
+    ip_network('fe80::/10'),
+    ip_network('fec0::/10'),
+    ip_network('ff00::/8'),
+]
+BOGONS = BOGONS_IP4.copy()
+BOGONS.extend(BOGONS_IP6)
+
+# dns-resolving firewall-rule content
+DNS_RESOLVE_TIMEOUT = 1
+DNS_RESOLVE_THREADS = 50
+
+IPLIST_DOWNLOAD_TIMEOUT = 3
+IPLIST_COMMENT_CHARS = ['#', ';']
 
 
 class Proto(ABC):
@@ -33,10 +74,11 @@ class ProtoL3IP4IP6(ProtoL3):
     N = 'ip'
 
 
-PROTOS_L3 = [ProtoL3IP4, ProtoL3IP6]
+PROTOS_L3 = [ProtoL3IP4, ProtoL3IP6, ProtoL3IP4IP6]
 PROTO_L3_MAPPING = {
     ProtoL3IP4.N: ProtoL3IP4,
     ProtoL3IP6.N: ProtoL3IP6,
+    ProtoL3IP4IP6.N: ProtoL3IP4IP6,
 }
 
 class ProtoL4(Proto):
diff --git a/src/firewall_test/plugins/system/config.py b/src/firewall_test/plugins/system/config.py
@@ -1,9 +1,14 @@
 from plugins.system.system_linux_netfilter import SystemLinuxNetfilter
 from plugins.translate.netfilter.ruleset import NetfilterRuleset
 from plugins.translate.linux import LinuxRoutes, LinuxRouteRules, LinuxNetworkInterfaces
+from plugins.system.system_opnsense import SystemOPNsense
+from plugins.translate.opnsense.interfaces import OPNsenseNetworkInterfaces
+from plugins.translate.opnsense.routes import OPNsenseRoutes
+from plugins.translate.opnsense.ruleset import OPNsenseRuleset
 
 SYSTEM_MAPPING = {
     'linux_netfilter': SystemLinuxNetfilter,
+    'opnsense': SystemOPNsense,
 }
 
 COMPONENT_MAPPING = {
@@ -12,5 +17,10 @@
         'routes': LinuxRoutes,
         'route_rules': LinuxRouteRules,
         'ruleset': NetfilterRuleset,
-    }
+    },
+    SystemOPNsense: {
+        'nis': OPNsenseNetworkInterfaces,
+        'routes': OPNsenseRoutes,
+        'ruleset': OPNsenseRuleset,
+    },
 }
diff --git a/src/firewall_test/plugins/system/firewall_opnsense.py b/src/firewall_test/plugins/system/firewall_opnsense.py
@@ -0,0 +1,16 @@
+from simulator.packet import PacketIP
+from plugins.translate.abstract import Rule
+from plugins.system.abstract_rule_match import RuleMatcher, RuleMatchResult
+from plugins.translate.opnsense.rule import OPNsenseRule
+
+
+class RuleMatcherOPNsense(RuleMatcher):
+    def matches(self, packet: PacketIP, rule: Rule) -> RuleMatchResult:
+        """
+        :param packet: Packet to match
+        :param rule: Rule to check
+        :return: RuleMatchResult
+        """
+        opn_rule: OPNsenseRule = rule.raw
+        del opn_rule
+        return RuleMatchResult(False, None, None, None, None)
diff --git a/src/firewall_test/plugins/system/system_opnsense.py b/src/firewall_test/plugins/system/system_opnsense.py
@@ -0,0 +1,57 @@
+# pylint: disable=R0801
+
+from config import FlowInput, FlowOutput, FlowForward
+from plugins.system.abstract import FirewallSystem
+from plugins.system.firewall_opnsense import RuleMatcher, RuleMatcherOPNsense
+
+
+class SystemOPNsense(FirewallSystem):
+    ROUTE_STATIC = True
+    ROUTE_STATIC_RULES = False
+
+    SYSTEM_DROP_WAN_BOGONS = True  # todo: instance-specific => read from config (interfaces-wan-blockbogons)
+    SYSTEM_DROP_FORWARD = False
+
+    FIREWALL_ACTION_LAZY = True
+    FIREWALL_CT = True
+    FIREWALL_PRIO_LOWER_BETTER = True
+    FIREWALL_PRIO_TABLE_FULL = False
+    FIREWALL_DNAT = True
+    FIREWALL_SNAT = True
+
+    # see: https://docs.opnsense.org/manual/firewall.html#processing-order
+    FIREWALL_HOOKS = {
+        FlowInput: ['dnat', 'floating', 'interface_groups', 'interfaces'],
+        FlowForward: ['dnat', 'floating', 'interface_groups', 'interfaces', 'snat'],
+        FlowOutput: ['dnat', 'floating', 'interface_groups', 'interfaces', 'snat'],
+        'full': ['dnat', 'floating', 'interface_groups', 'interfaces', 'snat'],
+    }
+    FIREWALL_INGRESS = {
+        FlowInput: {'hook': 'prerouting', 'priority': 1000},
+        FlowForward: {'hook': 'prerouting', 'priority': 1000},
+        FlowOutput: {'hook': 'output', 'priority': -100},
+    }
+    _chain_dnat = {'hook': 'dnat', 'priority': 0}
+    _chain_snat = {'hook': 'snat', 'priority': 0}
+    FIREWALL_NAT = {
+        FlowInput: {
+            'dnat': _chain_dnat,
+        },
+        FlowForward: {
+            'dnat': _chain_dnat,
+            'snat': _chain_snat,
+        },
+        FlowOutput: {
+            'dnat': _chain_dnat,
+            'snat': _chain_snat,
+        },
+    }
+
+    @classmethod
+    def get_rule_matcher(cls) -> type[RuleMatcher]:
+        """
+        Property to return the system-specific rule-matcher (plugins.system.abstract_rule_match.RuleMatcher)
+
+        :return: RuleMatcher
+        """
+        return RuleMatcherOPNsense
diff --git a/src/firewall_test/plugins/translate/netfilter/elements.py b/src/firewall_test/plugins/translate/netfilter/elements.py
@@ -6,6 +6,8 @@
 from plugins.translate.netfilter.parts import RULE_ACTIONS, IGNORE_RULE_EXPRESSIONS, IGNORE_LEFT
 from utils.logger import log_warn
 
+# pylint: disable=R0801
+
 # for schema see: https://www.mankier.com/5/libnftables-json
 
 def translate_family(family: str) -> type[ProtoL3]:
diff --git a/src/firewall_test/plugins/translate/netfilter/ruleset.py b/src/firewall_test/plugins/translate/netfilter/ruleset.py
@@ -27,7 +27,7 @@ def get(self) -> Rule:
 
 class NetfilterChainOutput(Chain):
     def _validate_hooks(self):
-        assert self.hook is None or self.hook in SystemLinuxNetfilter.FIREWALL_HOOKS
+        assert self.hook is None or self.hook in SystemLinuxNetfilter.FIREWALL_HOOKS['full']
 
 
 class NetfilterChain(TranslatePluginChain):
diff --git a/src/firewall_test/plugins/translate/opnsense/__init__.py b/src/firewall_test/plugins/translate/opnsense/__init__.py
diff --git a/src/firewall_test/plugins/translate/opnsense/interfaces_test.py b/src/firewall_test/plugins/translate/opnsense/interfaces_test.py
@@ -30,6 +30,7 @@ def test_opnsense_nis():
 
         elif ni.name == 'opt5':
             assert ni.up
+            assert ni.desc == 'WAN2'
             is_ips = [str(ip) for ip in ni.ip4]
             is_ips.sort()
             want_ips = [
diff --git a/src/firewall_test/plugins/translate/opnsense/rule.py b/src/firewall_test/plugins/translate/opnsense/rule.py
@@ -0,0 +1,114 @@
+from ipaddress import IPv4Network, IPv6Network
+
+from config import ProtoL3IP4IP6, PROTOS_L3, PROTOS_L4
+
+# pylint: disable=R0801
+
+# pylint: disable=R0912,R0913,R0914,R0915,R0917
+class OPNsenseRule:
+    DIRECTION_IN = 'in'
+    DIRECTION_OUT = 'out'
+    DIRECTION_ANY = 'any'
+
+    def __init__(
+            self,
+            nr: int,
+            nis: list[str] = None,
+            ni_direction: str = None,
+            desc: str = None,
+            quick: bool = True,
+            ipprotocol: PROTOS_L3 = ProtoL3IP4IP6,
+            protocol: PROTOS_L4 = None,
+            source: list[(IPv4Network, IPv6Network)] = None,
+            destination: list[(IPv4Network, IPv6Network)] = None,
+            source_port: list[int] = None,
+            destination_port: list[int] = None,
+            source_invert: bool = False,
+            destination_invert: bool = False,
+            source_any: bool = False,
+            destination_any: bool = False,
+    ):
+        self.nr = nr
+        self.nis = nis
+        self.ni_direction = ni_direction
+        self.desc = desc
+        self.quick = quick
+        self.ipp = ipprotocol
+        self.proto = protocol
+        self.src = source
+        self.dst = destination
+        self.src_port = source_port
+        self.dst_port = destination_port
+        self.src_invert = source_invert
+        self.dst_invert = destination_invert
+        self.src_any = source_any
+        self.dst_any = destination_any
+
+    @property
+    def match_ni_in(self) -> bool:
+        if len(self.nis) == 0:
+            return False
+
+        if self.ni_direction in [self.DIRECTION_IN, self.DIRECTION_ANY]:
+            return True
+
+        return False
+
+    @property
+    def match_ni_out(self) -> bool:
+        if len(self.nis) == 0:
+            return False
+
+        if self.ni_direction in [self.DIRECTION_OUT, self.DIRECTION_ANY]:
+            return True
+
+        return False
+
+    @property
+    def match_ip_saddr(self) -> bool:
+        if self.src_any or (self.src is not None and len(self.src) > 0):
+            return True
+
+        return False
+
+    @property
+    def match_ip_daddr(self) -> bool:
+        if self.dst_any or (self.src is not None and len(self.src) > 0):
+            return True
+
+        return False
+
+    def get_match_types(self) -> (list[str], None):
+        match = []
+        if self.ipp is not None:
+            match.append('proto_l3')
+
+        if self.proto is not None:
+            match.append('proto_l4')
+
+        if self.match_ip_saddr:
+            match.append('ip_saddr')
+
+        if self.match_ip_daddr:
+            match.append('ip_daddr')
+
+        if self.match_ni_in:
+            match.append('ni_in')
+
+        if self.match_ni_out:
+            match.append('ni_out')
+
+        if self.src_port is not None and len(self.src_port) > 0:
+            match.append('src-port')
+
+        if self.dst_port is not None and len(self.dst_port) > 0:
+            match.append('dst-port')
+
+        if len(match) == 0:
+            match = None
+
+        return match
+
+    def __repr__(self) -> str:
+        desc = '' if self.desc is None else f' ({self.desc})'
+        return f"Rule: #{self.nr}{desc} | Matches: {self.get_match_types()}"
diff --git a/src/firewall_test/plugins/translate/opnsense/ruleset.py b/src/firewall_test/plugins/translate/opnsense/ruleset.py
diff --git a/src/firewall_test/plugins/translate/opnsense/ruleset_test.py b/src/firewall_test/plugins/translate/opnsense/ruleset_test.py
diff --git a/testdata/plugin_translate_opnsense_config.xml b/testdata/plugin_translate_opnsense_config.xml
