From 991ed2c6ac56a1e36b6dc66afe3382800dac7908 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:17:22 +0200 Subject: [PATCH 01/10] test: add reversible firewall write lifecycle --- .../test_firewall_write_lifecycle.py | 545 ++++++++++++++++++ 1 file changed, 545 insertions(+) create mode 100644 tests/integration/test_firewall_write_lifecycle.py diff --git a/tests/integration/test_firewall_write_lifecycle.py b/tests/integration/test_firewall_write_lifecycle.py new file mode 100644 index 0000000..5292771 --- /dev/null +++ b/tests/integration/test_firewall_write_lifecycle.py @@ -0,0 +1,545 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import os +from collections.abc import Mapping +from typing import Any + +import pytest + +from nr2301 import NR2301Client + + +if os.environ.get("NR2301_WRITE_INTEGRATION") != "1": + pytest.skip( + "physical Firewall/NAT write tests require NR2301_WRITE_INTEGRATION=1", + allow_module_level=True, + ) + + +pytestmark = pytest.mark.integration + +_PF_FIELDS = ("name", "mac", "local_port", "wan_port") +_PT_FIELDS = ("name", "trigger_port", "start_port", "end_port") +_SYNTHETIC_IPS = ("203.0.113.77", "203.0.113.78", "203.0.113.79") +_SYNTHETIC_PORTS = ("65500:65500", "65499:65499", "65498:65498") +_SYNTHETIC_URLS = ( + "sdk-firewall-test.invalid", + "sdk-firewall-test-2.invalid", + "sdk-firewall-test-3.invalid", +) +_SYNTHETIC_MACS = ( + "02:00:00:00:00:FA", + "02:00:00:00:00:F9", + "02:00:00:00:00:F8", +) + + +@pytest.fixture(scope="module") +def router(): + password = os.environ.get("NR2301_PASSWORD") + if not password: + pytest.skip("NR2301_PASSWORD is required for physical-router integration tests") + + with NR2301Client( + os.environ.get("NR2301_URL", "http://zyxel.home"), + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=10.0, + ) as client: + client.login() + yield client + + +def _mapping(value: Any, label: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + pytest.fail(f"{label} did not return an object") + return value + + +def _firewall(response: Mapping[str, Any], label: str) -> Mapping[str, Any]: + return _mapping(response.get("firewall"), f"{label}.firewall") + + +def _binary(value: Any, label: str) -> int: + if isinstance(value, bool): + pytest.fail(f"{label} returned a boolean instead of 0/1") + try: + numeric = int(value) + except (TypeError, ValueError): + pytest.fail(f"{label} did not return 0/1") + if numeric not in {0, 1}: + pytest.fail(f"{label} returned unsupported value {numeric!r}") + return numeric + + +def _flag(response: Mapping[str, Any], key: str, label: str) -> int: + return _binary(_firewall(response, label).get(key), f"{label}.{key}") + + +def _vpn_state(router) -> tuple[int, int, int]: + response = router.firewall.vpn_passthrough(timeout=5.0) + return ( + _binary(response.get("pptp"), "vpn.pptp"), + _binary(response.get("l2tp"), "vpn.l2tp"), + _binary(response.get("ipsec"), "vpn.ipsec"), + ) + + +def _indexed_state( + response: Mapping[str, Any], + *, + fields: tuple[str, ...], + limit: int, + label: str, +) -> tuple[int, tuple[tuple[str, ...], ...]]: + settings = _mapping(response.get("settings"), f"{label}.settings") + enable = _binary(settings.get("enable"), f"{label}.settings.enable") + raw_items = settings.get("items") + if raw_items is None: + raw_items = [] + if not isinstance(raw_items, list): + pytest.fail(f"{label}.settings.items did not return a list") + + slots = [tuple("" for _ in fields) for _ in range(limit)] + seen: set[int] = set() + for raw in raw_items: + item = _mapping(raw, f"{label}.item") + index_value = item.get("index") + if isinstance(index_value, bool): + pytest.fail(f"{label} returned invalid boolean index") + try: + index = int(index_value) + except (TypeError, ValueError): + pytest.fail(f"{label} returned invalid index") + if not 0 <= index < limit: + pytest.fail(f"{label} returned out-of-range index {index}") + if index in seen: + pytest.fail(f"{label} returned duplicate index {index}") + seen.add(index) + slots[index] = tuple(str(item.get(field) or "") for field in fields) + + return enable, tuple(slots) + + +def _items_from_slots( + slots: tuple[tuple[str, ...], ...], + fields: tuple[str, ...], +) -> list[dict[str, object]]: + items: list[dict[str, object]] = [] + for index, values in enumerate(slots): + if not any(values): + continue + items.append( + {"index": index, **{field: value for field, value in zip(fields, values)}} + ) + return items + + +def _rule_state( + response: Mapping[str, Any], + *, + value_key: str, + label: str, +) -> tuple[str, ...]: + raw_list = _firewall(response, label).get("list") + if not isinstance(raw_list, list): + pytest.fail(f"{label}.firewall.list did not return a list") + + slots = ["0"] * 10 + seen: set[int] = set() + for raw in raw_list: + item = _mapping(raw, f"{label}.item") + index_value = item.get("index") + if isinstance(index_value, bool): + pytest.fail(f"{label} returned invalid boolean index") + try: + index = int(index_value) + except (TypeError, ValueError): + pytest.fail(f"{label} returned invalid index") + if not 0 <= index < 10: + pytest.fail(f"{label} returned out-of-range index {index}") + if index in seen: + pytest.fail(f"{label} returned duplicate index {index}") + seen.add(index) + value = item.get(value_key) + slots[index] = "0" if value in {None, "", "0"} else str(value) + return tuple(slots) + + +def _url_values(raw_items: Any, label: str) -> tuple[str, ...]: + if raw_items is None: + raw_items = [] + if not isinstance(raw_items, list): + pytest.fail(f"{label} did not return a list") + + slots = [""] * 10 + seen: set[int] = set() + for raw in raw_items: + item = _mapping(raw, f"{label}.item") + index_value = item.get("index") + if isinstance(index_value, bool): + pytest.fail(f"{label} returned invalid boolean index") + try: + index = int(index_value) + except (TypeError, ValueError): + pytest.fail(f"{label} returned invalid index") + if not 0 <= index < 10: + pytest.fail(f"{label} returned out-of-range index {index}") + if index in seen: + pytest.fail(f"{label} returned duplicate index {index}") + seen.add(index) + slots[index] = str(item.get("value") or "") + return tuple(slots) + + +def _url_state(router) -> tuple[str, tuple[str, ...], tuple[str, ...]]: + response = router.firewall.url_filter(timeout=5.0) + settings = _mapping(response.get("settings"), "url_filter.settings") + mode = settings.get("mode") + if mode not in {"disable", "blacklist", "whitelist"}: + pytest.fail(f"url_filter returned unsupported mode {mode!r}") + return ( + str(mode), + _url_values(settings.get("black_items"), "url_filter.black_items"), + _url_values(settings.get("white_items"), "url_filter.white_items"), + ) + + +def _dmz_destination(router) -> str: + response = router.firewall.dmz_info(timeout=5.0) + value = _firewall(response, "dmz_info").get("dmz_dest_ip") + if not isinstance(value, str): + pytest.fail("dmz_info did not return string dmz_dest_ip") + return value + + +def _snapshot(router) -> dict[str, object]: + return { + "dmz_enabled": 1 + - _flag(router.firewall.disable_info(timeout=5.0), "dmz_disable", "dmz"), + "dmz_destination": _dmz_destination(router), + "vpn": _vpn_state(router), + "admin_from_wan": _flag( + router.firewall.admin_from_wan(timeout=5.0), + "admin_from_wan_enable", + "admin_from_wan", + ), + "ping_from_wan": _flag( + router.firewall.ping_from_wan(timeout=5.0), + "ping_from_wan_enable", + "ping_from_wan", + ), + "port_forward": _indexed_state( + router.firewall.port_forward(timeout=5.0), + fields=_PF_FIELDS, + limit=5, + label="port_forward", + ), + "port_trigger": _indexed_state( + router.firewall.port_trigger(timeout=5.0), + fields=_PT_FIELDS, + limit=10, + label="port_trigger", + ), + "url_filter": _url_state(router), + "ip_filter_rules": _rule_state( + router.firewall.ip_filter(timeout=5.0), + value_key="ip", + label="ip_filter", + ), + "ip_filter_enabled": 1 + - _flag( + router.firewall.ip_filter_mode_state(timeout=5.0), + "ip_filter_disable", + "ip_filter_mode", + ), + "port_filter_rules": _rule_state( + router.firewall.port_filter(timeout=5.0), + value_key="port", + label="port_filter", + ), + "port_filter_enabled": 1 + - _flag( + router.firewall.port_filter_mode_state(timeout=5.0), + "port_filter_disable", + "port_filter_mode", + ), + "upnp_enabled": _flag( + router.firewall.upnp_state(timeout=5.0), + "upnp_enable", + "upnp", + ), + } + + +def _choose_unique(candidates: tuple[str, ...], existing: set[str]) -> str: + for candidate in candidates: + if candidate not in existing: + return candidate + pytest.fail("all reserved synthetic candidates are already present") + + +def _mutate_slot_values( + original: tuple[str, ...], + *, + candidates: tuple[str, ...], +) -> tuple[str, ...]: + values = list(original) + target = next((i for i, value in enumerate(values) if value in {"", "0"}), len(values) - 1) + synthetic = _choose_unique(candidates, {value for value in values if value not in {"", "0"}}) + values[target] = synthetic + return tuple(values) + + +def _mutate_indexed_slots( + original: tuple[tuple[str, ...], ...], + *, + fields: tuple[str, ...], + kind: str, +) -> tuple[tuple[str, ...], ...]: + slots = list(original) + target = next((i for i, values in enumerate(slots) if not any(values)), len(slots) - 1) + + if kind == "port_forward": + existing_macs = {values[1] for values in slots if len(values) > 1 and values[1]} + mac = _choose_unique(_SYNTHETIC_MACS, existing_macs) + replacement = ("SDK-PF-PHYSICAL", mac, "65500", "65500") + elif kind == "port_trigger": + replacement = ("SDK-PT-PHYSICAL", "65500", "65501", "65501") + else: + raise AssertionError(f"unknown indexed mutation kind: {kind}") + + if len(replacement) != len(fields): + raise AssertionError("synthetic indexed rule shape mismatch") + slots[target] = replacement + return tuple(slots) + + +def _restore(router, original: Mapping[str, object]) -> None: + # Restore full stored lists before their enable switches. + ip_rules = tuple(original["ip_filter_rules"]) + router.firewall.replace_ip_filter_rules(list(ip_rules), timeout=10.0) + router.firewall.set_ip_filter_enabled(bool(original["ip_filter_enabled"]), timeout=10.0) + + port_rules = tuple(original["port_filter_rules"]) + router.firewall.replace_port_filter_rules(list(port_rules), timeout=10.0) + router.firewall.set_port_filter_enabled( + bool(original["port_filter_enabled"]), timeout=10.0 + ) + + pf_enable, pf_slots = original["port_forward"] + pf_items = _items_from_slots(pf_slots, _PF_FIELDS) + if pf_enable or pf_items: + router.firewall.set_port_forward(True, items=pf_items, timeout=10.0) + if not pf_enable: + router.firewall.set_port_forward(False, timeout=10.0) + + pt_enable, pt_slots = original["port_trigger"] + pt_items = _items_from_slots(pt_slots, _PT_FIELDS) + if pt_enable or pt_items: + router.firewall.set_port_trigger(True, items=pt_items, timeout=10.0) + if not pt_enable: + router.firewall.set_port_trigger(False, timeout=10.0) + + url_mode, black_items, white_items = original["url_filter"] + if url_mode == "blacklist": + router.firewall.set_url_filter("blacklist", items=list(black_items), timeout=10.0) + elif url_mode == "whitelist": + router.firewall.set_url_filter("whitelist", items=list(white_items), timeout=10.0) + else: + # The physical test mutates only the blacklist when starting disabled, + # so restore that stored list before returning to disabled mode. + router.firewall.set_url_filter("blacklist", items=list(black_items), timeout=10.0) + router.firewall.set_url_filter("disable", timeout=10.0) + + pptp, l2tp, ipsec = original["vpn"] + router.firewall.set_vpn_passthrough( + pptp=bool(pptp), l2tp=bool(l2tp), ipsec=bool(ipsec), timeout=10.0 + ) + router.firewall.set_admin_from_wan(bool(original["admin_from_wan"]), timeout=10.0) + router.firewall.set_ping_from_wan(bool(original["ping_from_wan"]), timeout=10.0) + router.firewall.set_upnp_enabled(bool(original["upnp_enabled"]), timeout=10.0) + router.firewall.set_dmz_enabled(bool(original["dmz_enabled"]), timeout=10.0) + + +def test_firewall_write_lifecycle_and_exact_restore(router): + original = _snapshot(router) + + try: + router.firewall.set_dmz_enabled(not bool(original["dmz_enabled"]), timeout=10.0) + assert _snapshot(router)["dmz_enabled"] == 1 - int(original["dmz_enabled"]) + print("FIREWALL_DMZ_ENABLE_WRITE changed=True readback=True", flush=True) + + pptp, l2tp, ipsec = original["vpn"] + expected_vpn = (1 - pptp, l2tp, ipsec) + router.firewall.set_vpn_passthrough( + pptp=bool(expected_vpn[0]), + l2tp=bool(expected_vpn[1]), + ipsec=bool(expected_vpn[2]), + timeout=10.0, + ) + assert _vpn_state(router) == expected_vpn + print("FIREWALL_VPN_PASSTHROUGH_WRITE changed=True readback=True", flush=True) + + router.firewall.set_admin_from_wan( + not bool(original["admin_from_wan"]), timeout=10.0 + ) + assert ( + _flag( + router.firewall.admin_from_wan(timeout=5.0), + "admin_from_wan_enable", + "admin_from_wan", + ) + == 1 - int(original["admin_from_wan"]) + ) + print("FIREWALL_ADMIN_FROM_WAN_WRITE changed=True readback=True", flush=True) + + router.firewall.set_ping_from_wan( + not bool(original["ping_from_wan"]), timeout=10.0 + ) + assert ( + _flag( + router.firewall.ping_from_wan(timeout=5.0), + "ping_from_wan_enable", + "ping_from_wan", + ) + == 1 - int(original["ping_from_wan"]) + ) + print("FIREWALL_PING_FROM_WAN_WRITE changed=True readback=True", flush=True) + + mutated_ip = _mutate_slot_values( + tuple(original["ip_filter_rules"]), candidates=_SYNTHETIC_IPS + ) + router.firewall.replace_ip_filter_rules(list(mutated_ip), timeout=10.0) + assert ( + _rule_state( + router.firewall.ip_filter(timeout=5.0), + value_key="ip", + label="ip_filter", + ) + == mutated_ip + ) + print("FIREWALL_IP_FILTER_RULE_WRITE changed=True readback=True", flush=True) + + router.firewall.set_ip_filter_enabled( + not bool(original["ip_filter_enabled"]), timeout=10.0 + ) + assert ( + 1 + - _flag( + router.firewall.ip_filter_mode_state(timeout=5.0), + "ip_filter_disable", + "ip_filter_mode", + ) + == 1 - int(original["ip_filter_enabled"]) + ) + print("FIREWALL_IP_FILTER_MODE_WRITE changed=True readback=True", flush=True) + + mutated_port = _mutate_slot_values( + tuple(original["port_filter_rules"]), candidates=_SYNTHETIC_PORTS + ) + router.firewall.replace_port_filter_rules(list(mutated_port), timeout=10.0) + assert ( + _rule_state( + router.firewall.port_filter(timeout=5.0), + value_key="port", + label="port_filter", + ) + == mutated_port + ) + print("FIREWALL_PORT_FILTER_RULE_WRITE changed=True readback=True", flush=True) + + router.firewall.set_port_filter_enabled( + not bool(original["port_filter_enabled"]), timeout=10.0 + ) + assert ( + 1 + - _flag( + router.firewall.port_filter_mode_state(timeout=5.0), + "port_filter_disable", + "port_filter_mode", + ) + == 1 - int(original["port_filter_enabled"]) + ) + print("FIREWALL_PORT_FILTER_MODE_WRITE changed=True readback=True", flush=True) + + pf_enable, pf_slots = original["port_forward"] + mutated_pf = _mutate_indexed_slots( + pf_slots, fields=_PF_FIELDS, kind="port_forward" + ) + router.firewall.set_port_forward( + True, items=_items_from_slots(mutated_pf, _PF_FIELDS), timeout=10.0 + ) + assert _indexed_state( + router.firewall.port_forward(timeout=5.0), + fields=_PF_FIELDS, + limit=5, + label="port_forward", + ) == (1, mutated_pf) + print("FIREWALL_PORT_FORWARD_WRITE changed=True readback=True", flush=True) + + pt_enable, pt_slots = original["port_trigger"] + mutated_pt = _mutate_indexed_slots( + pt_slots, fields=_PT_FIELDS, kind="port_trigger" + ) + router.firewall.set_port_trigger( + True, items=_items_from_slots(mutated_pt, _PT_FIELDS), timeout=10.0 + ) + assert _indexed_state( + router.firewall.port_trigger(timeout=5.0), + fields=_PT_FIELDS, + limit=10, + label="port_trigger", + ) == (1, mutated_pt) + print("FIREWALL_PORT_TRIGGER_WRITE changed=True readback=True", flush=True) + + url_mode, black_items, white_items = original["url_filter"] + if url_mode == "whitelist": + mutated_urls = _mutate_slot_values( + tuple(white_items), candidates=_SYNTHETIC_URLS + ) + router.firewall.set_url_filter( + "whitelist", items=list(mutated_urls), timeout=10.0 + ) + assert _url_state(router)[0:1] == ("whitelist",) + assert _url_state(router)[2] == mutated_urls + else: + mutated_urls = _mutate_slot_values( + tuple(black_items), candidates=_SYNTHETIC_URLS + ) + router.firewall.set_url_filter( + "blacklist", items=list(mutated_urls), timeout=10.0 + ) + assert _url_state(router)[0:2] == ("blacklist", mutated_urls) + print("FIREWALL_URL_FILTER_WRITE changed=True readback=True", flush=True) + + router.firewall.set_upnp_enabled( + not bool(original["upnp_enabled"]), timeout=10.0 + ) + assert ( + _flag( + router.firewall.upnp_state(timeout=5.0), + "upnp_enable", + "upnp", + ) + == 1 - int(original["upnp_enabled"]) + ) + print("FIREWALL_UPNP_WRITE changed=True readback=True", flush=True) + + finally: + _restore(router, original) + + final = _snapshot(router) + assert final == original + print( + "FIREWALL_FINAL" + " dmz_restored=True" + " vpn_restored=True" + " wan_controls_restored=True" + " filters_restored=True" + " nat_rules_restored=True" + " upnp_restored=True", + flush=True, + ) From a9176c8d6dc53c4f01a8c319cde341fc060e2ec9 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:18:04 +0200 Subject: [PATCH 02/10] test: add recoverable DMZ destination lifecycle --- .../test_firewall_dmz_destination.py | 236 ++++++++++++++++++ 1 file changed, 236 insertions(+) create mode 100644 tests/integration/test_firewall_dmz_destination.py diff --git a/tests/integration/test_firewall_dmz_destination.py b/tests/integration/test_firewall_dmz_destination.py new file mode 100644 index 0000000..ceda729 --- /dev/null +++ b/tests/integration/test_firewall_dmz_destination.py @@ -0,0 +1,236 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import ipaddress +import os +import time +from collections.abc import Mapping +from typing import Any + +import pytest + +from nr2301 import NR2301Client, NR2301Error + + +if os.environ.get("NR2301_DESTRUCTIVE_INTEGRATION") != "1": + pytest.skip( + "physical DMZ-destination recovery test requires " + "NR2301_DESTRUCTIVE_INTEGRATION=1", + allow_module_level=True, + ) + + +pytestmark = pytest.mark.integration + + +@pytest.fixture(scope="module") +def router(): + password = os.environ.get("NR2301_PASSWORD") + if not password: + pytest.skip("NR2301_PASSWORD is required for physical-router integration tests") + + with NR2301Client( + os.environ.get("NR2301_URL", "http://zyxel.home"), + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=15.0, + ) as client: + client.login() + yield client + + +def _mapping(value: Any, label: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + pytest.fail(f"{label} did not return an object") + return value + + +def _firewall(response: Mapping[str, Any], label: str) -> Mapping[str, Any]: + return _mapping(response.get("firewall"), f"{label}.firewall") + + +def _binary(value: Any, label: str) -> int: + if isinstance(value, bool): + pytest.fail(f"{label} returned boolean instead of 0/1") + try: + numeric = int(value) + except (TypeError, ValueError): + pytest.fail(f"{label} did not return 0/1") + if numeric not in {0, 1}: + pytest.fail(f"{label} returned unsupported value {numeric!r}") + return numeric + + +def _snapshot(router) -> dict[str, object]: + dmz_info = _firewall(router.firewall.dmz_info(timeout=5.0), "dmz_info") + dmz_disable = _firewall( + router.firewall.disable_info(timeout=5.0), "dmz_disable" + ).get("dmz_disable") + admin = _firewall( + router.firewall.admin_from_wan(timeout=5.0), "admin_from_wan" + ).get("admin_from_wan_enable") + ping = _firewall( + router.firewall.ping_from_wan(timeout=5.0), "ping_from_wan" + ).get("ping_from_wan_enable") + upnp = _firewall(router.firewall.upnp_state(timeout=5.0), "upnp").get( + "upnp_enable" + ) + vpn = router.firewall.vpn_passthrough(timeout=5.0) + + destination = dmz_info.get("dmz_dest_ip") + if not isinstance(destination, str): + pytest.fail("dmz_info did not return string dmz_dest_ip") + + return { + "dmz_destination": destination, + "dmz_enabled": 1 - _binary(dmz_disable, "dmz_disable"), + "admin_from_wan": _binary(admin, "admin_from_wan"), + "ping_from_wan": _binary(ping, "ping_from_wan"), + "upnp_enabled": _binary(upnp, "upnp"), + "vpn": ( + _binary(vpn.get("pptp"), "vpn.pptp"), + _binary(vpn.get("l2tp"), "vpn.l2tp"), + _binary(vpn.get("ipsec"), "vpn.ipsec"), + ), + } + + +def _snapshot_with_recovery(router, *, attempts: int = 90) -> dict[str, object]: + last_error: NR2301Error | None = None + for attempt in range(attempts): + try: + return _snapshot(router) + except NR2301Error as exc: + last_error = exc + try: + router.login() + except NR2301Error: + pass + if attempt + 1 < attempts: + time.sleep(1.0) + raise AssertionError( + "Firewall state did not become readable after config restore: " + f"{type(last_error).__name__ if last_error else 'unknown'}" + ) + + +def _synthetic_destination(router, original: str) -> str: + response = router.lan.address(timeout=5.0) + values = _mapping(response.get("router"), "router_get_lan_ip.router") + lan_ip = values.get("lan_ip") + netmask = values.get("lan_netmask") + if not isinstance(lan_ip, str) or not isinstance(netmask, str): + pytest.fail("router_get_lan_ip did not return LAN IPv4/netmask strings") + + try: + address = ipaddress.IPv4Address(lan_ip) + network = ipaddress.IPv4Network(f"{lan_ip}/{netmask}", strict=False) + except ValueError as exc: + pytest.fail(f"router_get_lan_ip returned invalid IPv4 settings: {exc}") + + # Prefer high usable addresses without enumerating the whole subnet. + for offset in range(1, 32): + candidate_int = int(network.broadcast_address) - offset + if candidate_int <= int(network.network_address): + break + candidate = ipaddress.IPv4Address(candidate_int) + text = str(candidate) + if candidate != address and text != original: + return text + + # The API contract has already physically accepted a documentation-only + # IPv4 value. This fallback is only for an unusually tiny LAN prefix. + if original != "192.0.2.10": + return "192.0.2.10" + return "192.0.2.11" + + +def test_dmz_destination_write_and_recovery(router): + original = _snapshot(router) + original_destination = str(original["dmz_destination"]) + synthetic = _synthetic_destination(router, original_destination) + + # Keep a secret-bearing backup only in memory. It is never printed, + # persisted by the test, or committed. + backup = router.maintenance.download_config_backup(timeout=30.0) + assert backup + + restore_via_backup = not bool(original_destination.strip()) + backup_used = False + + print( + "FIREWALL_DMZ_DESTINATION_PREP" + f" original_empty={restore_via_backup}" + " backup_ready=True", + flush=True, + ) + + try: + if not bool(original["dmz_enabled"]): + router.firewall.set_dmz_enabled(True, timeout=10.0) + + router.firewall.set_dmz_destination(synthetic, timeout=10.0) + current = _snapshot(router) + assert current["dmz_destination"] == synthetic + + print( + "FIREWALL_DMZ_DESTINATION_WRITE changed=True readback=True", + flush=True, + ) + + finally: + if restore_via_backup: + result = router.maintenance.restore_config_backup( + backup, + action_timeout=30.0, + recovery_attempts=150, + recovery_delay=1.0, + recovery_timeout=4.0, + initial_delay=2.0, + ) + backup_used = True + print( + "FIREWALL_DMZ_DESTINATION_RESTORE" + " method=config_backup" + f" outage_observed={result.get('outage_observed')}" + f" reboot_evidence={result.get('reboot_evidence')}", + flush=True, + ) + else: + try: + router.firewall.set_dmz_destination( + original_destination, timeout=10.0 + ) + router.firewall.set_dmz_enabled( + bool(original["dmz_enabled"]), timeout=10.0 + ) + print( + "FIREWALL_DMZ_DESTINATION_RESTORE method=setter", + flush=True, + ) + except NR2301Error: + result = router.maintenance.restore_config_backup( + backup, + action_timeout=30.0, + recovery_attempts=150, + recovery_delay=1.0, + recovery_timeout=4.0, + initial_delay=2.0, + ) + backup_used = True + print( + "FIREWALL_DMZ_DESTINATION_RESTORE" + " method=config_backup_fallback" + f" outage_observed={result.get('outage_observed')}" + f" reboot_evidence={result.get('reboot_evidence')}", + flush=True, + ) + + final = _snapshot_with_recovery(router) if backup_used else _snapshot(router) + assert final == original + print( + "FIREWALL_DMZ_DESTINATION_FINAL restored=True" + f" backup_used={backup_used}", + flush=True, + ) From 0169a9ab54c4104dc44ad4c8283f788bf575114a Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:32:18 +0200 Subject: [PATCH 03/10] test: compare Port Forward MAC case semantically --- .../integration/test_firewall_write_lifecycle.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_firewall_write_lifecycle.py b/tests/integration/test_firewall_write_lifecycle.py index 5292771..15e2731 100644 --- a/tests/integration/test_firewall_write_lifecycle.py +++ b/tests/integration/test_firewall_write_lifecycle.py @@ -123,6 +123,19 @@ def _indexed_state( return enable, tuple(slots) + +def _normalize_indexed_slots( + slots: tuple[tuple[str, ...], ...], + fields: tuple[str, ...], +) -> tuple[tuple[str, ...], ...]: + normalized: list[tuple[str, ...]] = [] + for values in slots: + row = [] + for field, value in zip(fields, values): + row.append(value.lower() if field == "mac" else value) + normalized.append(tuple(row)) + return tuple(normalized) + def _items_from_slots( slots: tuple[tuple[str, ...], ...], fields: tuple[str, ...], @@ -477,7 +490,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): fields=_PF_FIELDS, limit=5, label="port_forward", - ) == (1, mutated_pf) + ) == (1, _normalize_indexed_slots(mutated_pf, _PF_FIELDS)) print("FIREWALL_PORT_FORWARD_WRITE changed=True readback=True", flush=True) pt_enable, pt_slots = original["port_trigger"] From 1425f83a0516c06eb7f9f47f4bcc3857170a5417 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:49:57 +0200 Subject: [PATCH 04/10] test: recover transient URL filter readback stalls --- .../test_firewall_write_lifecycle.py | 57 ++++++++++++++++--- 1 file changed, 50 insertions(+), 7 deletions(-) diff --git a/tests/integration/test_firewall_write_lifecycle.py b/tests/integration/test_firewall_write_lifecycle.py index 15e2731..2d79179 100644 --- a/tests/integration/test_firewall_write_lifecycle.py +++ b/tests/integration/test_firewall_write_lifecycle.py @@ -3,12 +3,13 @@ from __future__ import annotations import os +import time from collections.abc import Mapping from typing import Any import pytest -from nr2301 import NR2301Client +from nr2301 import NR2301Client, NR2301Error if os.environ.get("NR2301_WRITE_INTEGRATION") != "1": @@ -207,8 +208,12 @@ def _url_values(raw_items: Any, label: str) -> tuple[str, ...]: return tuple(slots) -def _url_state(router) -> tuple[str, tuple[str, ...], tuple[str, ...]]: - response = router.firewall.url_filter(timeout=5.0) +def _url_state( + router, + *, + timeout: float = 10.0, +) -> tuple[str, tuple[str, ...], tuple[str, ...]]: + response = router.firewall.url_filter(timeout=timeout) settings = _mapping(response.get("settings"), "url_filter.settings") mode = settings.get("mode") if mode not in {"disable", "blacklist", "whitelist"}: @@ -220,6 +225,44 @@ def _url_state(router) -> tuple[str, tuple[str, ...], tuple[str, ...]]: ) +def _url_state_with_recovery( + router, + *, + attempts: int = 5, + timeout: float = 10.0, + delay: float = 1.0, +) -> tuple[str, tuple[str, ...], tuple[str, ...]]: + last_error: NR2301Error | None = None + for attempt in range(1, attempts + 1): + try: + state = _url_state(router, timeout=timeout) + if attempt > 1: + print( + "FIREWALL_URL_FILTER_READBACK_RECOVERED" + f" attempt={attempt}" + f" previous_error={type(last_error).__name__ if last_error else None}", + flush=True, + ) + return state + except NR2301Error as exc: + last_error = exc + print( + "FIREWALL_URL_FILTER_READBACK_RETRY" + f" attempt={attempt}" + f" error={type(exc).__name__}", + flush=True, + ) + try: + router.login() + except NR2301Error: + pass + if attempt < attempts and delay: + time.sleep(delay) + + assert last_error is not None + raise last_error + + def _dmz_destination(router) -> str: response = router.firewall.dmz_info(timeout=5.0) value = _firewall(response, "dmz_info").get("dmz_dest_ip") @@ -256,7 +299,7 @@ def _snapshot(router) -> dict[str, object]: limit=10, label="port_trigger", ), - "url_filter": _url_state(router), + "url_filter": _url_state_with_recovery(router), "ip_filter_rules": _rule_state( router.firewall.ip_filter(timeout=5.0), value_key="ip", @@ -516,8 +559,8 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.set_url_filter( "whitelist", items=list(mutated_urls), timeout=10.0 ) - assert _url_state(router)[0:1] == ("whitelist",) - assert _url_state(router)[2] == mutated_urls + assert _url_state_with_recovery(router)[0:1] == ("whitelist",) + assert _url_state_with_recovery(router)[2] == mutated_urls else: mutated_urls = _mutate_slot_values( tuple(black_items), candidates=_SYNTHETIC_URLS @@ -525,7 +568,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.set_url_filter( "blacklist", items=list(mutated_urls), timeout=10.0 ) - assert _url_state(router)[0:2] == ("blacklist", mutated_urls) + assert _url_state_with_recovery(router)[0:2] == ("blacklist", mutated_urls) print("FIREWALL_URL_FILTER_WRITE changed=True readback=True", flush=True) router.firewall.set_upnp_enabled( From e517a1ed269a941ad7416673ead28372f3eb11ae Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:00:58 +0200 Subject: [PATCH 05/10] test: recover final Firewall snapshot after write sequences --- .../test_firewall_write_lifecycle.py | 94 +++++++++++++------ 1 file changed, 66 insertions(+), 28 deletions(-) diff --git a/tests/integration/test_firewall_write_lifecycle.py b/tests/integration/test_firewall_write_lifecycle.py index 2d79179..432e36d 100644 --- a/tests/integration/test_firewall_write_lifecycle.py +++ b/tests/integration/test_firewall_write_lifecycle.py @@ -80,7 +80,7 @@ def _flag(response: Mapping[str, Any], key: str, label: str) -> int: def _vpn_state(router) -> tuple[int, int, int]: - response = router.firewall.vpn_passthrough(timeout=5.0) + response = router.firewall.vpn_passthrough(timeout=timeout) return ( _binary(response.get("pptp"), "vpn.pptp"), _binary(response.get("l2tp"), "vpn.l2tp"), @@ -263,73 +263,111 @@ def _url_state_with_recovery( raise last_error -def _dmz_destination(router) -> str: - response = router.firewall.dmz_info(timeout=5.0) +def _dmz_destination(router, *, timeout: float = 10.0) -> str: + response = router.firewall.dmz_info(timeout=timeout) value = _firewall(response, "dmz_info").get("dmz_dest_ip") if not isinstance(value, str): pytest.fail("dmz_info did not return string dmz_dest_ip") return value -def _snapshot(router) -> dict[str, object]: +def _snapshot(router, *, timeout: float = 10.0) -> dict[str, object]: return { "dmz_enabled": 1 - - _flag(router.firewall.disable_info(timeout=5.0), "dmz_disable", "dmz"), - "dmz_destination": _dmz_destination(router), + - _flag(router.firewall.disable_info(timeout=timeout), "dmz_disable", "dmz"), + "dmz_destination": _dmz_destination(router, timeout=timeout), "vpn": _vpn_state(router), "admin_from_wan": _flag( - router.firewall.admin_from_wan(timeout=5.0), + router.firewall.admin_from_wan(timeout=timeout), "admin_from_wan_enable", "admin_from_wan", ), "ping_from_wan": _flag( - router.firewall.ping_from_wan(timeout=5.0), + router.firewall.ping_from_wan(timeout=timeout), "ping_from_wan_enable", "ping_from_wan", ), "port_forward": _indexed_state( - router.firewall.port_forward(timeout=5.0), + router.firewall.port_forward(timeout=timeout), fields=_PF_FIELDS, limit=5, label="port_forward", ), "port_trigger": _indexed_state( - router.firewall.port_trigger(timeout=5.0), + router.firewall.port_trigger(timeout=timeout), fields=_PT_FIELDS, limit=10, label="port_trigger", ), - "url_filter": _url_state_with_recovery(router), + "url_filter": _url_state(router, timeout=timeout), "ip_filter_rules": _rule_state( - router.firewall.ip_filter(timeout=5.0), + router.firewall.ip_filter(timeout=timeout), value_key="ip", label="ip_filter", ), "ip_filter_enabled": 1 - _flag( - router.firewall.ip_filter_mode_state(timeout=5.0), + router.firewall.ip_filter_mode_state(timeout=timeout), "ip_filter_disable", "ip_filter_mode", ), "port_filter_rules": _rule_state( - router.firewall.port_filter(timeout=5.0), + router.firewall.port_filter(timeout=timeout), value_key="port", label="port_filter", ), "port_filter_enabled": 1 - _flag( - router.firewall.port_filter_mode_state(timeout=5.0), + router.firewall.port_filter_mode_state(timeout=timeout), "port_filter_disable", "port_filter_mode", ), "upnp_enabled": _flag( - router.firewall.upnp_state(timeout=5.0), + router.firewall.upnp_state(timeout=timeout), "upnp_enable", "upnp", ), } + +def _snapshot_with_recovery( + router, + *, + attempts: int = 6, + timeout: float = 10.0, + delay: float = 1.0, +) -> dict[str, object]: + last_error: NR2301Error | None = None + for attempt in range(1, attempts + 1): + try: + snapshot = _snapshot(router, timeout=timeout) + if attempt > 1: + print( + "FIREWALL_SNAPSHOT_RECOVERED" + f" attempt={attempt}" + f" previous_error={type(last_error).__name__ if last_error else None}", + flush=True, + ) + return snapshot + except NR2301Error as exc: + last_error = exc + print( + "FIREWALL_SNAPSHOT_RETRY" + f" attempt={attempt}" + f" error={type(exc).__name__}", + flush=True, + ) + try: + router.login() + except NR2301Error: + pass + if attempt < attempts and delay: + time.sleep(delay) + + assert last_error is not None + raise last_error + def _choose_unique(candidates: tuple[str, ...], existing: set[str]) -> str: for candidate in candidates: if candidate not in existing: @@ -421,11 +459,11 @@ def _restore(router, original: Mapping[str, object]) -> None: def test_firewall_write_lifecycle_and_exact_restore(router): - original = _snapshot(router) + original = _snapshot_with_recovery(router) try: router.firewall.set_dmz_enabled(not bool(original["dmz_enabled"]), timeout=10.0) - assert _snapshot(router)["dmz_enabled"] == 1 - int(original["dmz_enabled"]) + assert _snapshot_with_recovery(router)["dmz_enabled"] == 1 - int(original["dmz_enabled"]) print("FIREWALL_DMZ_ENABLE_WRITE changed=True readback=True", flush=True) pptp, l2tp, ipsec = original["vpn"] @@ -444,7 +482,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.admin_from_wan(timeout=5.0), + router.firewall.admin_from_wan(timeout=timeout), "admin_from_wan_enable", "admin_from_wan", ) @@ -457,7 +495,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.ping_from_wan(timeout=5.0), + router.firewall.ping_from_wan(timeout=timeout), "ping_from_wan_enable", "ping_from_wan", ) @@ -471,7 +509,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.replace_ip_filter_rules(list(mutated_ip), timeout=10.0) assert ( _rule_state( - router.firewall.ip_filter(timeout=5.0), + router.firewall.ip_filter(timeout=timeout), value_key="ip", label="ip_filter", ) @@ -485,7 +523,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): assert ( 1 - _flag( - router.firewall.ip_filter_mode_state(timeout=5.0), + router.firewall.ip_filter_mode_state(timeout=timeout), "ip_filter_disable", "ip_filter_mode", ) @@ -499,7 +537,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.replace_port_filter_rules(list(mutated_port), timeout=10.0) assert ( _rule_state( - router.firewall.port_filter(timeout=5.0), + router.firewall.port_filter(timeout=timeout), value_key="port", label="port_filter", ) @@ -513,7 +551,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): assert ( 1 - _flag( - router.firewall.port_filter_mode_state(timeout=5.0), + router.firewall.port_filter_mode_state(timeout=timeout), "port_filter_disable", "port_filter_mode", ) @@ -529,7 +567,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): True, items=_items_from_slots(mutated_pf, _PF_FIELDS), timeout=10.0 ) assert _indexed_state( - router.firewall.port_forward(timeout=5.0), + router.firewall.port_forward(timeout=timeout), fields=_PF_FIELDS, limit=5, label="port_forward", @@ -544,7 +582,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): True, items=_items_from_slots(mutated_pt, _PT_FIELDS), timeout=10.0 ) assert _indexed_state( - router.firewall.port_trigger(timeout=5.0), + router.firewall.port_trigger(timeout=timeout), fields=_PT_FIELDS, limit=10, label="port_trigger", @@ -576,7 +614,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.upnp_state(timeout=5.0), + router.firewall.upnp_state(timeout=timeout), "upnp_enable", "upnp", ) @@ -587,7 +625,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): finally: _restore(router, original) - final = _snapshot(router) + final = _snapshot_with_recovery(router) assert final == original print( "FIREWALL_FINAL" From 05720db75d1ced3d31b56a4439d480c5b2bc34a4 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:01:32 +0200 Subject: [PATCH 06/10] test: fix Firewall recovery timeout scoping --- .../test_firewall_write_lifecycle.py | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/tests/integration/test_firewall_write_lifecycle.py b/tests/integration/test_firewall_write_lifecycle.py index 432e36d..997e0e7 100644 --- a/tests/integration/test_firewall_write_lifecycle.py +++ b/tests/integration/test_firewall_write_lifecycle.py @@ -79,7 +79,7 @@ def _flag(response: Mapping[str, Any], key: str, label: str) -> int: return _binary(_firewall(response, label).get(key), f"{label}.{key}") -def _vpn_state(router) -> tuple[int, int, int]: +def _vpn_state(router, *, timeout: float = 10.0) -> tuple[int, int, int]: response = router.firewall.vpn_passthrough(timeout=timeout) return ( _binary(response.get("pptp"), "vpn.pptp"), @@ -276,7 +276,7 @@ def _snapshot(router, *, timeout: float = 10.0) -> dict[str, object]: "dmz_enabled": 1 - _flag(router.firewall.disable_info(timeout=timeout), "dmz_disable", "dmz"), "dmz_destination": _dmz_destination(router, timeout=timeout), - "vpn": _vpn_state(router), + "vpn": _vpn_state(router, timeout=timeout), "admin_from_wan": _flag( router.firewall.admin_from_wan(timeout=timeout), "admin_from_wan_enable", @@ -482,7 +482,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.admin_from_wan(timeout=timeout), + router.firewall.admin_from_wan(timeout=10.0), "admin_from_wan_enable", "admin_from_wan", ) @@ -495,7 +495,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.ping_from_wan(timeout=timeout), + router.firewall.ping_from_wan(timeout=10.0), "ping_from_wan_enable", "ping_from_wan", ) @@ -509,7 +509,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.replace_ip_filter_rules(list(mutated_ip), timeout=10.0) assert ( _rule_state( - router.firewall.ip_filter(timeout=timeout), + router.firewall.ip_filter(timeout=10.0), value_key="ip", label="ip_filter", ) @@ -523,7 +523,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): assert ( 1 - _flag( - router.firewall.ip_filter_mode_state(timeout=timeout), + router.firewall.ip_filter_mode_state(timeout=10.0), "ip_filter_disable", "ip_filter_mode", ) @@ -537,7 +537,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): router.firewall.replace_port_filter_rules(list(mutated_port), timeout=10.0) assert ( _rule_state( - router.firewall.port_filter(timeout=timeout), + router.firewall.port_filter(timeout=10.0), value_key="port", label="port_filter", ) @@ -551,7 +551,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): assert ( 1 - _flag( - router.firewall.port_filter_mode_state(timeout=timeout), + router.firewall.port_filter_mode_state(timeout=10.0), "port_filter_disable", "port_filter_mode", ) @@ -567,7 +567,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): True, items=_items_from_slots(mutated_pf, _PF_FIELDS), timeout=10.0 ) assert _indexed_state( - router.firewall.port_forward(timeout=timeout), + router.firewall.port_forward(timeout=10.0), fields=_PF_FIELDS, limit=5, label="port_forward", @@ -582,7 +582,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): True, items=_items_from_slots(mutated_pt, _PT_FIELDS), timeout=10.0 ) assert _indexed_state( - router.firewall.port_trigger(timeout=timeout), + router.firewall.port_trigger(timeout=10.0), fields=_PT_FIELDS, limit=10, label="port_trigger", @@ -614,7 +614,7 @@ def test_firewall_write_lifecycle_and_exact_restore(router): ) assert ( _flag( - router.firewall.upnp_state(timeout=timeout), + router.firewall.upnp_state(timeout=10.0), "upnp_enable", "upnp", ) From 608de10e3e8fa08b302e3c49e18c90e54307cb82 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Fri, 25 Sep 2026 07:26:07 +0200 Subject: [PATCH 07/10] test: apply Firewall recovery policy to DMZ lifecycle --- .../test_firewall_dmz_destination.py | 27 +++++++++++-------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/tests/integration/test_firewall_dmz_destination.py b/tests/integration/test_firewall_dmz_destination.py index ceda729..d861579 100644 --- a/tests/integration/test_firewall_dmz_destination.py +++ b/tests/integration/test_firewall_dmz_destination.py @@ -62,21 +62,21 @@ def _binary(value: Any, label: str) -> int: return numeric -def _snapshot(router) -> dict[str, object]: - dmz_info = _firewall(router.firewall.dmz_info(timeout=5.0), "dmz_info") +def _snapshot(router, *, timeout: float = 10.0) -> dict[str, object]: + dmz_info = _firewall(router.firewall.dmz_info(timeout=timeout), "dmz_info") dmz_disable = _firewall( - router.firewall.disable_info(timeout=5.0), "dmz_disable" + router.firewall.disable_info(timeout=timeout), "dmz_disable" ).get("dmz_disable") admin = _firewall( - router.firewall.admin_from_wan(timeout=5.0), "admin_from_wan" + router.firewall.admin_from_wan(timeout=timeout), "admin_from_wan" ).get("admin_from_wan_enable") ping = _firewall( - router.firewall.ping_from_wan(timeout=5.0), "ping_from_wan" + router.firewall.ping_from_wan(timeout=timeout), "ping_from_wan" ).get("ping_from_wan_enable") - upnp = _firewall(router.firewall.upnp_state(timeout=5.0), "upnp").get( + upnp = _firewall(router.firewall.upnp_state(timeout=timeout), "upnp").get( "upnp_enable" ) - vpn = router.firewall.vpn_passthrough(timeout=5.0) + vpn = router.firewall.vpn_passthrough(timeout=timeout) destination = dmz_info.get("dmz_dest_ip") if not isinstance(destination, str): @@ -96,11 +96,16 @@ def _snapshot(router) -> dict[str, object]: } -def _snapshot_with_recovery(router, *, attempts: int = 90) -> dict[str, object]: +def _snapshot_with_recovery( + router, + *, + attempts: int = 12, + timeout: float = 10.0, +) -> dict[str, object]: last_error: NR2301Error | None = None for attempt in range(attempts): try: - return _snapshot(router) + return _snapshot(router, timeout=timeout) except NR2301Error as exc: last_error = exc try: @@ -147,7 +152,7 @@ def _synthetic_destination(router, original: str) -> str: def test_dmz_destination_write_and_recovery(router): - original = _snapshot(router) + original = _snapshot_with_recovery(router) original_destination = str(original["dmz_destination"]) synthetic = _synthetic_destination(router, original_destination) @@ -227,7 +232,7 @@ def test_dmz_destination_write_and_recovery(router): flush=True, ) - final = _snapshot_with_recovery(router) if backup_used else _snapshot(router) + final = _snapshot_with_recovery(router) assert final == original print( "FIREWALL_DMZ_DESTINATION_FINAL restored=True" From 2369a87923e922f81d22839d14700e794c6ee741 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Fri, 25 Sep 2026 07:47:19 +0200 Subject: [PATCH 08/10] fix: validate DMZ destination IPv4 --- src/nr2301/namespaces/firewall.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/nr2301/namespaces/firewall.py b/src/nr2301/namespaces/firewall.py index a2596fc..32b5d31 100644 --- a/src/nr2301/namespaces/firewall.py +++ b/src/nr2301/namespaces/firewall.py @@ -2,6 +2,7 @@ from __future__ import annotations +import ipaddress from collections.abc import Mapping, Sequence from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast @@ -270,18 +271,26 @@ def set_dmz_enabled( def set_dmz_destination( self, destination: str, *, timeout: float | None = None ) -> dict[str, Any]: - """Write a non-empty DMZ destination. + """Write a verified IPv4 DMZ destination. Empty-string clear/delete is deliberately rejected because no such - NR2301 WebUI contract has been verified. + NR2301 WebUI contract has been verified. Raw getter values are not + guaranteed to be IPv4 addresses and must not be replayed blindly. """ - if not destination.strip(): + text = destination.strip() + if not text: raise ValueError("DMZ destination clear/delete is not a verified contract") + try: + normalized = str(ipaddress.IPv4Address(text)) + except ipaddress.AddressValueError as exc: + raise ValueError( + "DMZ destination must be a syntactically valid IPv4 address" + ) from exc return self._client.call( "firewall", "fw_edit_dmz_entry", - data={"dmz_dest_ip": destination}, + data={"dmz_dest_ip": normalized}, timeout=timeout, ) From a257b09976e27ce8a3aac0a0183a8c3bf77ac419 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Fri, 25 Sep 2026 07:47:34 +0200 Subject: [PATCH 09/10] test: classify DMZ restore by IPv4 validity --- .../test_firewall_dmz_destination.py | 28 ++++++++++++++++--- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/tests/integration/test_firewall_dmz_destination.py b/tests/integration/test_firewall_dmz_destination.py index d861579..1dc7bdc 100644 --- a/tests/integration/test_firewall_dmz_destination.py +++ b/tests/integration/test_firewall_dmz_destination.py @@ -120,6 +120,14 @@ def _snapshot_with_recovery( ) +def _is_valid_ipv4(value: str) -> bool: + try: + ipaddress.IPv4Address(value.strip()) + except ipaddress.AddressValueError: + return False + return True + + def _synthetic_destination(router, original: str) -> str: response = router.lan.address(timeout=5.0) values = _mapping(response.get("router"), "router_get_lan_ip.router") @@ -161,12 +169,14 @@ def test_dmz_destination_write_and_recovery(router): backup = router.maintenance.download_config_backup(timeout=30.0) assert backup - restore_via_backup = not bool(original_destination.strip()) + original_setter_restorable = _is_valid_ipv4(original_destination) + restore_via_backup = not original_setter_restorable backup_used = False print( "FIREWALL_DMZ_DESTINATION_PREP" - f" original_empty={restore_via_backup}" + f" original_setter_restorable={original_setter_restorable}" + f" backup_restore_required={restore_via_backup}" " backup_ready=True", flush=True, ) @@ -203,6 +213,7 @@ def test_dmz_destination_write_and_recovery(router): flush=True, ) else: + setter_restore_ok = False try: router.firewall.set_dmz_destination( original_destination, timeout=10.0 @@ -210,11 +221,20 @@ def test_dmz_destination_write_and_recovery(router): router.firewall.set_dmz_enabled( bool(original["dmz_enabled"]), timeout=10.0 ) + restored = _snapshot_with_recovery(router) + setter_restore_ok = ( + restored["dmz_destination"] == original_destination + and restored["dmz_enabled"] == original["dmz_enabled"] + ) + except NR2301Error: + setter_restore_ok = False + + if setter_restore_ok: print( - "FIREWALL_DMZ_DESTINATION_RESTORE method=setter", + "FIREWALL_DMZ_DESTINATION_RESTORE method=setter readback=True", flush=True, ) - except NR2301Error: + else: result = router.maintenance.restore_config_backup( backup, action_timeout=30.0, From fcb8bf1e75fe56c88f74190419c266f19d03df7a Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Fri, 25 Sep 2026 07:47:43 +0200 Subject: [PATCH 10/10] test: reject invalid DMZ destination values --- tests/test_firewall_write_helpers.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/test_firewall_write_helpers.py b/tests/test_firewall_write_helpers.py index 80bf7c9..fe4a8ee 100644 --- a/tests/test_firewall_write_helpers.py +++ b/tests/test_firewall_write_helpers.py @@ -116,6 +116,27 @@ def test_dmz_destination_write_rejects_unverified_clear(): assert session.calls == [] +def test_dmz_destination_write_rejects_invalid_raw_getter_value(): + client, session = authenticated_client() + + with pytest.raises(ValueError, match="valid IPv4"): + client.firewall.set_dmz_destination("192.168.") + + assert session.calls == [] + + +def test_dmz_destination_write_normalizes_valid_ipv4_text(): + payload = {"firewall": {"setting_response": "OK"}} + client, session = authenticated_client(payload) + + assert client.firewall.set_dmz_destination(" 192.0.2.10 ") == payload + assert_single_post( + session, + "fw_edit_dmz_entry", + {"dmz_dest_ip": "192.0.2.10"}, + ) + + def test_dmz_destination_write_uses_verified_edit_method(): payload = {"firewall": {"setting_response": "OK"}} client, session = authenticated_client(payload)