Merge pull request #64 from NethServer/issue671

mwan: fix object support

src/nethsec/mwan/__init__.py

@@ -44,7 +44,6 @@ def _is_valid_dst(e_uci: EUci, database_id: str):
     """
     Validate the given object for destination.
     Destination objects can be only:
-    - domain set
     - dhcp reservation
     - dns domain
     - vpn user
@@ -60,7 +59,7 @@ def _is_valid_dst(e_uci: EUci, database_id: str):
     if objects.is_host_set(e_uci, database_id):
         return objects.is_singleton_host_set(e_uci, database_id, allow_cidr=True)
 
-    return objects.is_domain_set(e_uci, database_id) or objects.is_host(e_uci, database_id) or objects.is_domain(e_uci, database_id) or objects.is_vpn_user(e_uci, database_id)
+    return objects.is_host(e_uci, database_id) or objects.is_domain(e_uci, database_id) or objects.is_vpn_user(e_uci, database_id)
 
 def __generate_metric(e_uci: EUci) -> int:
     """
@@ -685,15 +684,11 @@ def update_rules(e_uci: EUci):
     for rule in utils.get_all_by_type(e_uci, 'mwan3', 'rule'):
         ns_src = e_uci.get('mwan3', rule, 'ns_src', default=None)
         ns_dst = e_uci.get('mwan3', rule, 'ns_dst', default=None)
+        # both ns_src and ns_dst should be a singleton
         if ns_src:
             e_uci.set('mwan3', rule, 'src_ip', objects.get_object_ip(e_uci, ns_src))
-        if ns_dst: # this can be only a domain set
-            id = ns_dst.split('/')[1]
-            ipsets = objects.get_domain_set_ipsets(e_uci, id)
-            e_uci.set('mwan3', rule, 'ipset', f"{ipsets['firewall']} dst")
-            try:
-                e_uci.delete('mwan3', rule, 'dest_ip')
-            except:
-                pass
+        if ns_dst:
+            # domain sets are not supported because mwan3 ipset functionality is broken on 23.05.
+            e_uci.set('mwan3', rule, 'dest_ip', objects.get_object_ip(e_uci, ns_dst))
 
     e_uci.save('mwan3')

src/nethsec/objects/__init__.py

@@ -827,20 +827,23 @@ def get_info(uci, database_id):
 
     Returns:
         a dictionary with the following fields:
+
         - `database`: the database of the object
         - `id`: the id of the object
         - `name`: the name of the object
         - `type`: the type of the object
+        - `family`: IP family, like `ipv4` or `ipv6`
     """
     try:
         database, id = database_id.split('/')
         type = uci.get(database, id)
         name = uci.get(database, id, 'name', default=None)
+        family = uci.get(database, id, 'family', default='ipv4')
         if not name:
             name = uci.get(database, id, 'label', default=None)
             if not name:
                 name = id
-        return {'database': database, 'id': id, 'name': name, 'type': type}
+        return {'database': database, 'id': id, 'name': name, 'type': type, 'family': family}
     except:
         return None
 

tests/test_mwan.py

@@ -337,10 +337,15 @@ def test_store_rule(e_uci, mocker):
     assert e_uci.get('mwan3', 'ns_rule_1', 'sticky') == '1'
 
     domain_id = objects.add_domain_set(e_uci, "mydomainset6", "ipv4", ["test1.com", "test2.com"])
-    id = mwan.store_rule(e_uci, 'r_with_obj', 'ns_default', 'udp', ns_src="dhcp/ns_host_mwan", ns_dst=f"objects/{domain_id}")
+    with pytest.raises(ValueError):
+        mwan.store_rule(e_uci, 'r_with_obj', 'ns_default', 'udp', ns_src="dhcp/ns_host_mwan", ns_dst=f"objects/{domain_id}")
+
+    hostset_id = objects.add_host_set(e_uci, "myhostset", "ipv4", ["192.168.1.1"])
+    id = mwan.store_rule(e_uci, 'r_with_obj', 'ns_default', 'udp', ns_src="dhcp/ns_host_mwan", ns_dst=f"objects/{hostset_id}")
+
     id = id.split('.')[1]
     assert e_uci.get('mwan3', id, 'ns_src') == "dhcp/ns_host_mwan"
-    assert e_uci.get('mwan3', id, 'ns_dst') == f"objects/{domain_id}"
+    assert e_uci.get('mwan3', id, 'ns_dst') == f"objects/{hostset_id}"
     with pytest.raises(ValueError):
         mwan.store_rule(e_uci, 'rule_with_obj', 'ns_default', 'udp', ns_src="dhcp/ns_host_mwan", ns_dst="objects/invalid") 
     with pytest.raises(ValueError):
@@ -594,9 +599,11 @@ def test_update_rules(e_uci, mocker):
         }
     ])
     domain_id = objects.add_domain_set(e_uci, "mydomainset7", "ipv4", ["test1.com", "test2.com"])
-    id = mwan.store_rule(e_uci, 'r_with_obj', 'ns_cool_policy', 'udp', ns_src="dhcp/ns_domain_mwan", ns_dst=f"objects/{domain_id}")
+    with pytest.raises(ValidationError):
+        mwan.store_rule(e_uci, 'r_with_obj', 'ns_cool_policy', 'udp', ns_src="dhcp/ns_domain_mwan", ns_dst=f"objects/{domain_id}")
+
+    id = mwan.store_rule(e_uci, 'r_with_obj', 'ns_cool_policy', 'udp', ns_src="dhcp/ns_domain_mwan", ns_dst=f"dhcp/ns_host_mwan")
     id = id.split('.')[1]
-    ipsets = objects.get_domain_set_ipsets(e_uci, domain_id)
     mwan.update_rules(e_uci)
     assert e_uci.get('mwan3', id, 'src_ip') == '7.8.9.1'
-    assert e_uci.get('mwan3', id, 'ipset') == f"{ipsets['firewall']} dst"
+    assert e_uci.get('mwan3', id, 'dest_ip') == '192.168.100.5'

tests/test_objects.py

@@ -22,6 +22,11 @@
 	option name 'h1'
 	option family 'ipv4'
 	list ipaddr 'dhcp/ns_112233'
+
+config host 'ns_a7439990'
+	option name 'h1v6'
+	option family 'ipv6'
+	list ipaddr 'fd:618c:d80a:dc82:2380:54c7:7a17:1013'
 """
 
 firewall_db = """
@@ -243,7 +248,7 @@ def test_is_used_host_set(u):
 
 def test_list_host_sets(u):
     sets = objects.list_host_sets(u)
-    assert len(sets) == 9
+    assert len(sets) == 10
 
 def test_is_singleton_host_set(u):
     id1 = objects.add_host_set(u, "myhostset", "ipv4", ["1.2.3.4", "5.6.7.8"])
@@ -362,7 +367,13 @@ def test_get_reference_info(u):
     assert ref['type'] == 'host'
     assert ref['name'] == 'host2'
     assert ref['id'] == 'ns_8dcab636'
+    assert ref['family'] == 'ipv4'
     assert objects.get_info(u, "unknown") == None
+    ref2 = objects.get_info(u, "objects/ns_a7439990")
+    assert ref2['type'] == 'host'
+    assert ref2['name'] == 'h1v6'
+    assert ref2['id'] == 'ns_a7439990'
+    assert ref2['family'] == 'ipv6'
 
 def test_is_object_with_invalid_is(u):
     assert objects.is_object_id(None) == False