adding basic translate-plugins for opnsense routes & network-interfaces

Author: superstes

docs/source/_static/img/plugin-opnsense-backup.png

@@ -1,5 +1,11 @@
 .. _plugins_fw_opnsense:
 
+.. |export_backup| image:: ../_static/img/plugin-opnsense-backup.png
+   :class: wiki-img
+
+.. |export_network| image:: ../_static/img/plugin-opnsense-export.png
+   :class: wiki-img
+
 .. include:: ../_include/head.rst
 
 ===================
@@ -11,16 +17,29 @@ Firewall - OPNsense
 Config Export
 #############
 
-1. `Download a Config-Backup <https://docs.opnsense.org/manual/backups.html>`_
+1. `Download a Config-Backup <https://docs.opnsense.org/manual/backups.html>`_ (referenced as :code:`config.xml`)
+
+    |export_backup|
 
-2. `Supply the runtime routes manually <https://docs.opnsense.org/manual/routes.html#status>`_ or `query them via API <https://docs.opnsense.org/development/api/core/diagnostics.html#id6>`_
+2. Download the current network status via the WebUI: :code:`Interfaces - Overview - Download Buttom` (referenced as :code:`network.json`)
+
+    |export_network|
 
 ----
 
 Run
 ###
 
-.. include:: ../_include/warn_wip.rst
+Here is an example on how to run supply the exported config:
+
+.. code-block:: bash
+
+    ftf-cli --firewall-system 'linux_netfilter' \
+            --file-ruleset 'config.xml' \
+            --file-interfaces 'network.json' \
+            --file-routes 'network.json' \
+            --src-ip 172.17.11.5 \
+            --dst-ip 1.1.1.1
 
 ----
 

docs/source/_static/img/plugin-opnsense-export.png

@@ -10,6 +10,7 @@
 DEFAULT_ROUTE_IP4 = ip_network('0.0.0.0/0')
 DEFAULT_ROUTE_IP6 = ip_network('::/0')
 DEFAULT_ROUTES = [DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6]
+LINK_LOCAL_IP6 = ip_network('fe80::/64')
 
 
 class Proto(ABC):

docs/source/plugins/firewall_opnsense.rst

@@ -32,13 +32,12 @@ def validate(self):
 class StaticRoute(TranslateOutput):
     # pylint: disable=W0622
     def __init__(
-        self, table: str, net: str, scope: str, type: str, gw: str = None, src_pref: str = None,
+        self, table: str, net: str, scope: str, gw: str = None, src_pref: str = None,
         ni: str = None, metric: int = None,
     ):
         self.table = table
         self.net = net
         self.scope = scope
-        self.type = type
         self.gw = gw
         self.src_pref = src_pref
         self.ni = ni
@@ -48,7 +47,7 @@ def __init__(
             self.src_pref = ip_address(self.src_pref)
 
     def __repr__(self) -> str:
-        return f'ROUTE: Network {self.net} in Table {self.table} via {self.ni} {self.gw} metric {self.metric} type {self.type}'
+        return f'ROUTE: Network {self.net} in Table {self.table} via {self.ni} {self.gw} metric {self.metric}'
 
     def dump(self) -> dict:
         gw, net = None, None
@@ -66,7 +65,6 @@ def dump(self) -> dict:
             'table': self.table,
             'net': net,
             'scope': self.scope,
-            'type': self.type,
             'gw': gw,
             'src_pref': self.src_pref,
             'ni': self.ni,
@@ -84,8 +82,7 @@ def validate(self):
             assert isinstance(r['src_pref'], (IPv4Address, IPv6Address))
 
         assert r['table'] in ['default', 'main', 'local', 'test']
-        assert r['type'] in ['default', 'local', 'broadcast', 'multicast']
-        assert r['scope'] in ['link', 'host', 'global', 'remote']
+        assert r['scope'] in ['link', 'local', 'global']
 
     def ip_count(self) -> int:
         cidr = int(str(self.net).split('/')[1])

src/firewall_test/config.py

@@ -1,10 +1,12 @@
-from ipaddress import ip_address, ip_network, IPv4Address
 from json import loads as json_loads
+from ipaddress import ip_address, ip_network, IPv4Address
 
 from config import DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6, ProtoL3IP4, ProtoL3IP6
 from plugins.translate.abstract import TranslatePluginStaticRoutes, TranslatePluginStaticRouteRules, \
     StaticRoute, StaticRouteRule, TranslatePluginNetworkInterfaces, NetworkInterface
 
+# pylint: disable=R0801
+
 
 class LinuxRouteRules(TranslatePluginStaticRouteRules):
     def __init__(self, raw: str):
@@ -48,18 +50,22 @@ def get(self) -> list[StaticRoute]:
         return [
             StaticRoute(**self._parse_route(r))
             for r in self.raw
+            if r.get('type', '') not in ['broadcast', 'multicast']
         ]
 
     @staticmethod
     def _parse_route(raw: dict) -> dict:
+        scope = raw.get('scope', 'global')
+        if scope == 'host':
+            scope = 'local'
+
         r = {
-            'scope': raw.get('scope', 'remote'),
+            'scope':scope,
             'ni': raw.get('dev', None),
             'metric': raw.get('metric', None),
             'src_pref': raw.get('prefsrc', None),
             'gw': raw.get('gateway', None),
             'table': raw.get('table', 'default'),
-            'type': raw.get('type', 'default'),
         }
 
         if raw.get('dst') == 'default':
@@ -92,14 +98,14 @@ def get(self) -> list[NetworkInterface]:
     @staticmethod
     def _parse_ni(raw: dict) -> dict:
         r = {
-            'name': raw.get('ifname'),
-            'mac': raw.get('address'),
+            'name': raw['ifname'],
+            'mac': raw['address'],
             ProtoL3IP4.N: [],
             ProtoL3IP6.N: [],
             'net4': [],
             'net6': [],
+            'up': raw.get('operstate') == 'UP' or raw['ifname'] == 'lo',
         }
-        r['up'] = raw.get('operstate') == 'UP' or r['name'] == 'lo'
 
         for info in raw.get('addr_info'):
             ip = ip_address(info['local'])

src/firewall_test/plugins/translate/abstract.py

@@ -7,10 +7,11 @@
 def test_linux_nis():
     from plugins.translate.linux import LinuxNetworkInterfaces
 
-    r = LinuxNetworkInterfaces(TESTDATA_NICS)
-    o = r.get()
+    nis = LinuxNetworkInterfaces(TESTDATA_NICS).get()
 
-    for ni in o:
+    assert len(nis) == 4
+
+    for ni in nis:
         ni.validate()
 
         if ni.name == 'lo':

src/firewall_test/plugins/translate/linux.py

@@ -0,0 +1,51 @@
+from ipaddress import ip_network
+
+from testdata_test import TESTDATA_FILE_ROUTES, TESTDATA_FILE_ROUTE_RULES
+
+from config import DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6, LINK_LOCAL_IP6
+
+with open(TESTDATA_FILE_ROUTES, 'r', encoding='utf-8') as f:
+    TESTDATA_ROUTES = f.read()
+
+with open(TESTDATA_FILE_ROUTE_RULES, 'r', encoding='utf-8') as f:
+    TESTDATA_RULES = f.read()
+
+
+def test_linux_rules():
+    from plugins.translate.linux import LinuxRouteRules
+
+    r = LinuxRouteRules(TESTDATA_RULES).get()
+
+    assert len(r) == 4
+
+    for rule in r:
+        rule.validate()
+
+        if rule.table == 'test':
+            assert rule.priority == 32765
+            assert len(rule.src) == 1
+            assert rule.src[0] == ip_network('172.18.0.0/16')
+
+
+def test_linux_routes():
+    from plugins.translate.linux import LinuxRoutes
+
+    r = LinuxRoutes(TESTDATA_ROUTES).get()
+
+    assert len(r) == 13
+
+    for route in r:
+        route.validate()
+
+        if route.net == DEFAULT_ROUTE_IP4:
+            assert str(route.gw) == '10.255.255.254'
+            assert route.scope == 'global'
+
+        elif route.net == DEFAULT_ROUTE_IP6:
+            assert str(route.gw) == 'fe80::7474:ceff:feb1:5347'
+            assert route.scope == 'global'
+
+        elif str(route.net) == '10.255.255.0/24':
+            assert route.gw is None
+            assert route.scope == 'link'
+            assert str(route.src_pref) == '10.255.255.48'

src/firewall_test/plugins/translate/linux_ni_test.py renamed to src/firewall_test/plugins/translate/linux_interfaces_test.py

@@ -0,0 +1,55 @@
+from json import loads as json_loads
+from ipaddress import ip_address, ip_network, IPv4Address
+
+from config import ProtoL3IP4, ProtoL3IP6
+from plugins.translate.abstract import NetworkInterface, TranslatePluginNetworkInterfaces
+
+# pylint: disable=R0801
+
+
+class OPNsenseNetworkInterfaces(TranslatePluginNetworkInterfaces):
+    def __init__(self, raw: str):
+        super().__init__(json_loads(raw))
+
+    def get(self) -> list[NetworkInterface]:
+        return [
+            NetworkInterface(**self._parse_ni(r))
+            for r in self.raw
+            if r['status'] != 'down'
+        ]
+
+    @staticmethod
+    def _parse_ni(raw: dict) -> dict:
+        r = {
+            'name': raw.get('identifier'),
+            'mac': raw.get('macaddr'),
+            ProtoL3IP4.N: [],
+            ProtoL3IP6.N: [],
+            'net4': [],
+            'net6': [],
+            'up': raw['status'] == 'up',
+        }
+
+        ips = raw.get('ipv4', [])
+        ips.extend(raw.get('ipv6', []))
+        for ip_cnf in ips:
+            ip_cidr = ip_cnf['ipaddr']
+            ip, cidr = ip_cidr.split('/', 1)
+            ip = ip_address(ip)
+            if cidr in ['32', '128']:
+                net = None
+
+            else:
+                net = ip_network(ip_cidr, strict=False)
+
+            if isinstance(ip, IPv4Address):
+                r[ProtoL3IP4.N].append(ip)
+                if net is not None:
+                    r['net4'].append(net)
+
+            else:
+                r[ProtoL3IP6.N].append(ip)
+                if net is not None:
+                    r['net6'].append(net)
+
+        return r