fix(python3-nethsec): avoiding object expasion if not needed

packages/python3-nethsec/src/nethsec/firewall/__init__.py

@@ -2047,36 +2047,35 @@ def update_firewall_rules(uci):
         keep_ipset = False
         ns_src = uci.get('firewall', section, 'ns_src', default=None)
         ns_dst = uci.get('firewall', section, 'ns_dst', default=None)
-        name = uci.get('firewall', section, 'name', default=None)
         if ns_src:
             if objects.is_domain_set(uci, ns_src):
                 keep_ipset = True
                 id = ns_src.split('/')[1]
                 ipsets = objects.get_domain_set_ipsets(uci, id)
-                uci.set('firewall', section, 'ipset', f"{ipsets['firewall']} src")
+                utils.set_if_changed(uci, 'firewall', section, 'ipset', f"{ipsets['firewall']} src")
                 try:
                     uci.delete('firewall', section, 'src_ip')
                 except:
                     pass
             else:
                 ipaddr = objects.get_object_ips(uci, ns_src)
                 if ipaddr:
-                    uci.set('firewall', section, 'src_ip', ipaddr)
+                    utils.set_if_changed(uci, 'firewall', section, 'src_ip', ipaddr)
 
         if ns_dst:
             if objects.is_domain_set(uci, ns_dst):
                 keep_ipset = True
                 id = ns_dst.split('/')[1]
                 ipsets = objects.get_domain_set_ipsets(uci, id)
-                uci.set('firewall', section, 'ipset', f"{ipsets['firewall']} dst")
+                utils.set_if_changed(uci, 'firewall', section, 'ipset', f"{ipsets['firewall']} dst")
                 try:
                     uci.delete('firewall', section, 'dest_ip')
                 except:
                     pass
             else:
                 ipaddr = objects.get_object_ips(uci, ns_dst)
                 if ipaddr:
-                    uci.set('firewall', section, 'dest_ip', ipaddr)
+                    utils.set_if_changed(uci, 'firewall', section, 'dest_ip', ipaddr)
 
         # delete ipset field if no domains are set
         if not keep_ipset:

packages/python3-nethsec/src/nethsec/objects/__init__.py

@@ -141,7 +141,8 @@ def get_object_ips(uci, database_id):
                 else:
                     ips.append(ip)
 
-    return list(set(ips))  # Convert the list to a set to remove duplicates, then convert it back to a list
+    # dict conversion into list, done so that this is sorted now
+    return list(dict.fromkeys(ips))
 
 def get_object_ip(uci, database_id):
     """

packages/python3-nethsec/src/nethsec/utils/__init__.py

@@ -127,6 +127,33 @@ def get_all_by_type(uci, config, utype):
     except:
         return None
 
+
+def set_if_changed(uci, config: str, section: str, option: str, value: str | list[str] | tuple[str, ...]) -> None:
+    '''
+    Drop-in replacement for uci.set that skips the write when the stored value already matches,
+    so UCI records no change
+
+    Arguments:
+      - uci -- EUci pointer
+      - config -- Configuration database name
+      - section -- Section name
+      - option -- Option name
+      - value -- Value to set, either a scalar or a list
+
+    Returns:
+      - Whatever uci.set returns
+    '''
+    if isinstance(value, (list, tuple)):
+        current = uci.get(config, section, option, list=True, default=None)
+        if current is not None and tuple(current) == tuple(value):
+            return None
+    else:
+        current = uci.get(config, section, option, default=None)
+        if current is not None and current == value:
+            return None
+    return uci.set(config, section, option, value)
+
+
 def get_device_name(hwaddr):
     '''
     Retrieve the physical device name given the MAC address

packages/python3-nethsec/tests/test_firewall.py

@@ -473,6 +473,38 @@ def u(tmp_path: pathlib.Path) -> EUci:
         fp.write(dpi_db)
     return EUci(confdir=tmp_path.as_posix())
 
+@pytest.fixture
+def u_saved(tmp_path: pathlib.Path) -> EUci:
+    # same fake dbs, but with a dedicated savedir so pending changes can be inspected
+    conf_dir = tmp_path.joinpath('conf')
+    conf_dir.mkdir()
+    save_dir = tmp_path.joinpath('save')
+    save_dir.mkdir()
+    with conf_dir.joinpath('firewall').open('w') as fp:
+        fp.write(firewall_db)
+        fp.write(zone_testing_db)
+    with conf_dir.joinpath('network').open('w') as fp:
+        fp.write(network_db)
+    with conf_dir.joinpath('templates').open('w') as fp:
+        fp.write(templates_db)
+    with conf_dir.joinpath('dhcp').open('w') as fp:
+        fp.write(dhcp_db)
+    with conf_dir.joinpath('netmap').open('w') as fp:
+        fp.write(netmap_db)
+    with conf_dir.joinpath('objects').open('w') as fp:
+        fp.write(objects_db)
+    with conf_dir.joinpath('users').open('w') as fp:
+        fp.write(user_db)
+    with conf_dir.joinpath('mwan3').open('w') as fp:
+        fp.write(mwan3_db)
+    with conf_dir.joinpath('dpi').open('w') as fp:
+        fp.write(dpi_db)
+    return EUci(confdir=conf_dir.as_posix(), savedir=save_dir.as_posix())
+
+def pending_changes(uci, config):
+    delta = pathlib.Path(uci.savedir()).joinpath(config)
+    return delta.read_text() if delta.exists() else ""
+
 def test_add_interface_to_zone(u):
     z1 = firewall.add_interface_to_zone(u, "interface1", "lan")
     assert z1 == 'lan1'
@@ -1188,6 +1220,36 @@ def test_update_firewall_rules_concurrency(u):
     objects.delete_host_set(u, host1)
     objects.delete_host_set(u, host2)
 
+def test_update_firewall_rules_is_idempotent(u_saved):
+    host1 = objects.add_host_set(u_saved, "hi1", "ipv4", ["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+    domain1 = objects.add_domain_set(u_saved, "di1", "ipv4", ["test1.com", "test2.com"])
+    hrid = firewall.add_rule(u_saved, "idemhost", "lan", [], "wan", [], [], [], "ACCEPT", "*",
+                             ns_src=f"objects/{host1}")
+    drid = firewall.add_rule(u_saved, "idemdomain", "lan", [], "wan", [], [], [], "ACCEPT", "*",
+                             ns_src=f"objects/{domain1}")
+    for config in ("firewall", "objects", "dhcp"):
+        u_saved.commit(config)
+
+    firewall.update_firewall_rules(u_saved)
+    u_saved.save("firewall")
+    changes = pending_changes(u_saved, "firewall")
+    assert f"firewall.{hrid}." not in changes
+    assert f"firewall.{drid}." not in changes
+
+def test_edit_rule_does_not_touch_other_object_rules(u_saved):
+    host1 = objects.add_host_set(u_saved, "he1", "ipv4", ["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+    orid = firewall.add_rule(u_saved, "withobject", "lan", [], "wan", [], [], [], "ACCEPT", "*",
+                             ns_src=f"objects/{host1}")
+    prid = firewall.add_rule(u_saved, "plainrule", "lan", ["192.168.1.1"], "wan", [], [], [], "ACCEPT", "*")
+    for config in ("firewall", "objects"):
+        u_saved.commit(config)
+
+    firewall.edit_rule(u_saved, prid, "plainrenamed", "lan", ["192.168.1.1"], "wan", [], [], [], "ACCEPT", "*")
+    u_saved.save("firewall")
+    changes = pending_changes(u_saved, "firewall")
+    assert f"firewall.{prid}." in changes
+    assert f"firewall.{orid}." not in changes
+
 def test_list_object_suggestions(u):
     obj = objects.list_objects(u)
     assert len(obj) == 9

packages/python3-nethsec/tests/test_objects.py

@@ -292,7 +292,8 @@ def test_get_object_ips(u):
     id0 = objects.add_host_set(u, "myhostset0", "ipv4", ["4.5.6.7"])
     id = objects.add_host_set(u, "myhostset", "ipv4", ["1.2.3.4", "dhcp/ns_8bec5896", "users/ns_user1", f"objects/{id0}"])
     ips = objects.get_object_ips(u, f"objects/{id}")
-    assert set(ips) == set(["1.2.3.4", "7.8.9.1", "10.10.10.22", "4.5.6.7"]) # check with set to ignore order
+    # this is ordered and must be ordered
+    assert ips == ["1.2.3.4", "7.8.9.1", "10.10.10.22", "4.5.6.7"]
 
 def test_is_domain_set(u):
     id = objects.add_domain_set(u, "mydomainset6", "ipv4", ["test1.com", "test2.com"])