fixes for opnsense ip-ranges and routes edge-case

Author: superstes

src/firewall_test/plugins/translate/opnsense/interfaces.py

@@ -35,7 +35,13 @@ def _parse_ni(raw: dict) -> dict:
         ips.extend(raw.get('ipv6', []))
         for ip_cnf in ips:
             ip_cidr = ip_cnf['ipaddr']
-            ip, cidr = ip_cidr.split('/', 1)
+            if ip_cidr.find('/') == -1:
+                ip = ip_cidr
+                cidr = str(ip_cnf.get('subnetbits', '32'))
+
+            else:
+                ip, cidr = ip_cidr.split('/', 1)
+
             ip = ip_address(ip)
             if cidr in ['32', '128']:
                 net = None

src/firewall_test/plugins/translate/opnsense/routes.py

@@ -33,7 +33,13 @@ def _parse_routes(raw: dict) -> list[dict]:
         ni = raw['identifier']
         gws = raw.get('gateways', [])
         routes = []
-        src_pref_ip4, src_pref_ip6 = raw['config'].get('ipaddr', None), raw['config'].get('ipaddrv6', None)
+        src_pref_ip4, src_pref_ip6 = None, None
+        if 'config' in raw:
+            src_pref_ip4 = raw['config'].get('ipaddr', None)
+            src_pref_ip6 = raw['config'].get('ipaddrv6', None)
+            if src_pref_ip6 is not None and src_pref_ip6.find(':') == -1:
+                src_pref_ip6 = None  # track6 ?!
+
         gw_ip4, gw_ip6 = None, None
         default_ip4 = False
 

src/firewall_test/plugins/translate/opnsense/ruleset.py

@@ -1,5 +1,5 @@
 from threading import Lock
-from ipaddress import ip_network
+from ipaddress import ip_network, summarize_address_range, ip_address
 import xml.etree.ElementTree as ET
 from xml.etree.ElementTree import Element
 
@@ -74,13 +74,22 @@ def _validate_hooks(self):
 
 
 def split_csv(value: str) -> list:
+    if value is None:
+        return []
+
     return [v for v in value.split(',') if v.strip() != '']
 
 
+def split_nlsv(value: list) -> list:
+    out = []
+    for v in value:
+        out.extend([v2.strip() for v2 in v.split('\n')])
+
+    return out
+
+
 def xml_to_dict(element: Element) -> dict:
-    out = {}
-    if hasattr(element, 'uuid'):
-        out['uuid'] = element.uuid
+    out = {'uuid': element.get('uuid', None)}
 
     for c in element:
         if len(list(c)) > 0:
@@ -160,10 +169,18 @@ def _parse_rule_address(self, value: str) -> (list, None):
 
         for v in values:
             if v in self.aliases:
+                if self.aliases[v] is None:
+                    log_warn('Firewall Plugin', f'Unable to parse rule-address: "{value}"')
+                    return None
+
                 out.extend(self.aliases[v])
                 continue
 
-            if f'{v}ip' in self.nis_ips:
+            if v in self.nis_ips:
+                out.extend(self.nis_ips[v])
+                continue
+
+            if v in self.nis_nets:
                 out.extend(self.nis_nets[v])
                 continue
 
@@ -182,9 +199,17 @@ def _parse_rule_network(self, value: str) -> (list, None):
 
         for v in values:
             if v in self.aliases:
+                if self.aliases[v] is None:
+                    log_warn('Firewall Plugin', f'Unable to parse rule-network: "{value}"')
+                    return None
+
                 out.extend(self.aliases[v])
                 continue
 
+            if v in self.nis_ips:
+                out.extend(self.nis_ips[v])
+                continue
+
             if v in self.nis_nets:
                 out.extend(self.nis_nets[v])
                 continue
@@ -258,8 +283,10 @@ def _parse_rules_old(self):
             else:
                 chain = self.chain_ni
 
-            build['uuid'] = rule.get('uuid', None)
             build['desc'] = rule.get('descr', None)
+            build['uuid'] = rule.get('uuid', None)
+            if build['uuid'] is None:
+                continue
 
             ### SEQUENCE ###
             nr += 1
@@ -278,6 +305,7 @@ def _parse_rules_old(self):
 
             ### RESOLVE NETWORK-INTERFACE GROUPS ###
             build['nis'] = []
+            build['ni_direction'] = rule.get('direction', 'any')
             for ni in nis:
                 if ni in self.ni_grp:
                     build['nis'].extend(self.ni_grp[ni])
@@ -376,18 +404,27 @@ def _parse_local_ips(self):
                 ip = p.get(ipp, None)
                 cidr = p.get(subnet, '32')
                 if ip is not None and ip.strip() != '':
-                    nets.append(ip_network(f'{ip}/{cidr}', strict=False))
-                    ip = ip_network(ip)
-                    self.local_ips.append(ip)
-                    ips.append(ip)
+                    try:
+                        nets.append(ip_network(f'{ip}/{cidr}', strict=False))
+
+                    except ValueError:
+                        log_warn('Firewall Plugin', f'Unable to parse IP: "{ip}/{cidr}"')
+
+                    try:
+                        ip = ip_network(ip)
+                        self.local_ips.append(ip)
+                        ips.append(ip)
+
+                    except ValueError:
+                        log_warn('Firewall Plugin', f'Unable to parse IP: "{ip}"')
 
-            self.nis_ips[ni.tag] = ips
+            self.nis_ips[f'{ni.tag}ip'] = ips
             self.nis_nets[ni.tag] = nets
 
         vips_raw = self.raw.getroot().find(XML_ELEMENT_VIPS)
         for vip in vips_raw:
             p = xml_to_dict(vip)
-            self.local_ips.append(ip_network(f"{p['subnet']}/{p['subnet_bits']}"))
+            self.local_ips.append(ip_network(p['subnet'], strict=False))
 
     ### ALIASES ###
 
@@ -446,7 +483,7 @@ def _resolve_alias_dns(self, aliases: dict):
             parallel=DNS_RESOLVE_THREADS,
         )
 
-    def _parse_alias_iplist_plain(self, url: str) -> list[str]:
+    def _parse_alias_iplist_plain(self, url: str) -> (list[str], None):
         content = []
         res = self._download_alias_iplist(url)
         if res is None:
@@ -479,6 +516,7 @@ def _parse_alias_iplist_plain(self, url: str) -> list[str]:
 
         if len(content) == 0:
             log_warn('Firewall Plugin', f'Alias-type "urltable" resulted in empty list: "{url}"')
+            return None
 
         return content
 
@@ -525,11 +563,16 @@ def _parse_aliases(self):
                 self.aliases[name] = ports
 
             elif t == 'network':
-                self.aliases[name] = [ip_network(n) for n in p['content']]
+                try:
+                    self.aliases[name] = [ip_network(n) for n in split_nlsv(p['content'])]
+
+                except ValueError:
+                    invalid = True
+                    possible_nested[name] = p['content']
 
             elif t == 'host':
                 content = []
-                for c in p['content']:
+                for c in split_nlsv(p['content']):
                     if valid_domain(c):
                         if c in self._dns_cache:
                             content.extend(self._dns_cache[c])
@@ -541,6 +584,15 @@ def _parse_aliases(self):
                 invalid = False
                 for c in content:
                     try:
+                        if c.find('-') != -1:
+                            ip_range1, ip_range2 = c.split('-', 1)
+                            range_nets = list(summarize_address_range(
+                                ip_address(ip_range1),
+                                ip_address(ip_range2),
+                            ))
+                            nets.extend(range_nets)
+                            continue
+
                         nets.append(ip_network(c))
 
                     except ValueError: