diff --git a/packages/python3-nethsec/src/nethsec/firewall/__init__.py b/packages/python3-nethsec/src/nethsec/firewall/__init__.py index 4adb67ba5..da4411af4 100644 --- a/packages/python3-nethsec/src/nethsec/firewall/__init__.py +++ b/packages/python3-nethsec/src/nethsec/firewall/__init__.py @@ -2047,13 +2047,12 @@ 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: @@ -2061,14 +2060,14 @@ def update_firewall_rules(uci): 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: @@ -2076,7 +2075,7 @@ def update_firewall_rules(uci): 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: diff --git a/packages/python3-nethsec/src/nethsec/objects/__init__.py b/packages/python3-nethsec/src/nethsec/objects/__init__.py index ece401749..c505ac70d 100644 --- a/packages/python3-nethsec/src/nethsec/objects/__init__.py +++ b/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): """ diff --git a/packages/python3-nethsec/src/nethsec/utils/__init__.py b/packages/python3-nethsec/src/nethsec/utils/__init__.py index c71955505..d892d4c6c 100644 --- a/packages/python3-nethsec/src/nethsec/utils/__init__.py +++ b/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 diff --git a/packages/python3-nethsec/tests/test_firewall.py b/packages/python3-nethsec/tests/test_firewall.py index 0920bcb31..63681520e 100644 --- a/packages/python3-nethsec/tests/test_firewall.py +++ b/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 diff --git a/packages/python3-nethsec/tests/test_objects.py b/packages/python3-nethsec/tests/test_objects.py index 4ef6a4a79..aa57ee00c 100644 --- a/packages/python3-nethsec/tests/test_objects.py +++ b/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"])