diff --git a/examples/capture_dhcp_static_contract.py b/examples/capture_dhcp_static_contract.py new file mode 100644 index 0000000..7c68536 --- /dev/null +++ b/examples/capture_dhcp_static_contract.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from urllib.parse import urljoin, urlsplit + +from nr2301 import NR2301Client + + +TARGETS = ( + "router_set_dhcp_static_ip", + "router_get_dhcp_static_ip", + "static", + "mapping", + "mac", + "index", + "toStringData", +) + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + out_dir = Path( + os.environ.get( + "NR2301_DHCP_SOURCE_OUT", + "dhcp_static_contract_capture", + ) + ).resolve() + out_dir.mkdir(parents=True, exist_ok=True) + + base = os.environ.get("NR2301_URL", "http://zyxel.home").rstrip("/") + origin = f"{urlsplit(base).scheme}://{urlsplit(base).netloc}" + + with NR2301Client( + base, + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=15.0, + ) as client: + client.login() + session = client.transport.session + + page_url = urljoin(base + "/", "html/set_dhcp.html") + response = session.get(page_url, timeout=15.0) + response.raise_for_status() + page = response.text + + script_urls = [] + for match in re.finditer( + r"""]+src\s*=\s*['"]([^'"]+)['"]""", + page, + re.IGNORECASE, + ): + url = urljoin(page_url, match.group(1)) + if urlsplit(url).netloc == urlsplit(origin).netloc: + script_urls.append(url) + + assets = [(page_url, page)] + for url in script_urls: + r = session.get(url, timeout=15.0) + r.raise_for_status() + assets.append((url, r.text)) + + report = {"assets": []} + for url, text in assets: + hits = [] + for target in TARGETS: + for match in re.finditer(re.escape(target), text, re.IGNORECASE): + start = max(0, match.start() - 2200) + end = min(len(text), match.end() + 4200) + snippet = text[start:end] + hits.append( + { + "target": target, + "start": start, + "end": end, + "snippet": snippet, + } + ) + if hits: + report["assets"].append({"url": url, "hits": hits}) + + report_path = out_dir / "report.json" + report_path.write_text( + json.dumps(report, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + print(f"DHCP_STATIC_SOURCE_REPORT = {report_path}") + print("DHCP_STATIC_SOURCE_CAPTURE = PASS") + + +if __name__ == "__main__": + main() diff --git a/examples/check_cleanup_dhcp_static_residue.py b/examples/check_cleanup_dhcp_static_residue.py new file mode 100644 index 0000000..3856a9c --- /dev/null +++ b/examples/check_cleanup_dhcp_static_residue.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import os +import time + +from nr2301 import NR2301Client, NR2301Error + + +SYNTHETIC_MACS = { + "02:00:00:00:00:fe", + "02:00:00:00:00:fd", + "02:00:00:00:00:fc", +} + + +def normalized_mac(value): + return str(value).replace("-", ":").lower() + + +def is_synthetic(item): + return normalized_mac(item["mac"]) in SYNTHETIC_MACS + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + 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() + current = client.lan.static_reservation_list(timeout=5.0) + + synthetic_present = any(is_synthetic(item) for item in current) + print( + "DHCP_STATIC_RESIDUE" + f" count={len(current)}" + f" synthetic_present={synthetic_present}" + ) + + if not synthetic_present: + print("DHCP_STATIC_RESIDUE_CLEANUP = NOT_NEEDED") + return + + if len(current) != 1: + raise RuntimeError( + "synthetic reservation is present together with other entries; " + "refusing automatic cleanup" + ) + + # The integration probe starts from an empty reservation table and + # uses only the reserved locally-administered synthetic MACs above. + # Clear only when one such test reservation is the sole current entry. + try: + client.multicall( + [ + { + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": []}, + "timeout": 30, + } + ], + timeout=30.0, + ) + except NR2301Error: + pass + + for attempt in range(30): + try: + after = client.lan.static_reservation_list(timeout=4.0) + if not after: + print("DHCP_STATIC_RESIDUE_CLEANUP = PASS") + return + except NR2301Error: + try: + client.login() + except NR2301Error: + pass + if attempt < 29: + time.sleep(1.0) + + raise RuntimeError("synthetic DHCP reservation residue could not be cleared") + + +if __name__ == "__main__": + main() diff --git a/examples/probe_work_mode_response.py b/examples/probe_work_mode_response.py new file mode 100644 index 0000000..c051ce5 --- /dev/null +++ b/examples/probe_work_mode_response.py @@ -0,0 +1,32 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +import os + +from nr2301 import NR2301Client + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + 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() + response = client.device.work_mode(timeout=5.0) + + print( + "WORK_MODE_RESPONSE = " + + json.dumps(response, ensure_ascii=True, sort_keys=True) + ) + + +if __name__ == "__main__": + main() diff --git a/src/nr2301/namespaces/__init__.py b/src/nr2301/namespaces/__init__.py index 4d9f25c..4327b8f 100644 --- a/src/nr2301/namespaces/__init__.py +++ b/src/nr2301/namespaces/__init__.py @@ -43,6 +43,7 @@ LANAddress, LANAddressResponse, LANNamespace, + StaticReservation, ) from .maintenance import MaintenanceNamespace, MaintenanceRecoveryResult, TimedRebootSettings from .mobile import ( @@ -215,6 +216,7 @@ "SMSSendResult", "SignalInfo", "SleepWaitTime", + "StaticReservation", "StatisticsNamespace", "TimedRebootSettings", "TrafficCounters", diff --git a/src/nr2301/namespaces/device.py b/src/nr2301/namespaces/device.py index 4ef3e69..8f9bd80 100644 --- a/src/nr2301/namespaces/device.py +++ b/src/nr2301/namespaces/device.py @@ -2,9 +2,10 @@ from __future__ import annotations +import time from typing import TYPE_CHECKING, Any, TypedDict, cast -from ..exceptions import APIError, ProtocolError +from ..exceptions import APIError, NR2301Error, ProtocolError, TransportError if TYPE_CHECKING: from ..client import NR2301Client @@ -86,8 +87,7 @@ class UILanguage(TypedDict, total=False): class WorkMode(TypedDict, total=False): - mode: str - result: int + work_mode: str class BatteryInfo(TypedDict, total=False): @@ -182,6 +182,94 @@ def work_mode(self, *, timeout: float | None = None) -> WorkMode: self._client.call("router", "router_get_work_mode", timeout=timeout), ) + def set_work_mode( + self, + mode: str, + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + force: bool = False, + ) -> WorkMode: + """Set router/bridge work mode through the verified multicall setter. + + Changing work mode can fundamentally alter addressing and management + reachability. A lost write response is therefore inconclusive; success + requires recovery plus exact getter read-back. + + `force=True` executes the setter even when the current mode already + matches, which is useful for explicit transport verification. + """ + + if mode not in {"router", "bridge"}: + raise ValueError("mode must be 'router' or 'bridge'") + if write_timeout <= 0: + raise ValueError("write_timeout must be greater than zero") + if recovery_attempts <= 0: + raise ValueError("recovery_attempts must be greater than zero") + if recovery_delay < 0: + raise ValueError("recovery_delay must not be negative") + if recovery_timeout <= 0: + raise ValueError("recovery_timeout must be greater than zero") + if not isinstance(force, bool): + raise TypeError("force must be a bool") + + current = self.work_mode(timeout=recovery_timeout) + current_mode = self._work_mode_code(current) + if current_mode == mode and not force: + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [ + { + "path": "router", + "method": "router_set_work_mode", + "data": {"work_mode": mode}, + "timeout": write_timeout, + } + ], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: str | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + verified = self.work_mode(timeout=recovery_timeout) + last_actual = self._work_mode_code(verified) + if last_actual == mode: + return verified + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": mode, + "actual": last_actual, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "router work-mode write could not be verified by exact read-back", + method_id="router/router_set_work_mode", + response=details, + ) + def set_ui_language( self, language: str, @@ -292,6 +380,16 @@ def set_sleep_wait_time( ) return verified + @staticmethod + def _work_mode_code(response: WorkMode) -> str: + value: Any = response.get("work_mode") + if not isinstance(value, str) or not value: + raise ProtocolError( + "router/router_get_work_mode did not return a non-empty " + "work_mode string" + ) + return value + @staticmethod def _ui_language_code(response: UILanguage) -> str: value: Any = response.get("language") diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index bd1949c..4e3f742 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -65,6 +65,12 @@ class LANAddressResponse(TypedDict, total=False): router: LANAddress +class StaticReservation(TypedDict): + index: int + mac: str + ip: str + + class DNSSettings(TypedDict): """Verified DNS subset of the combined DHCP settings object.""" @@ -161,6 +167,235 @@ def static_reservations(self, *, timeout: float | None = None) -> dict[str, Any] timeout=timeout, ) + def set_dhcp_settings( + self, + settings: Mapping[str, Any], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> DHCPSettings: + """Write the complete verified 12-field combined DHCP object. + + Start from `lan.dhcp()`, modify only intended fields, then pass the + complete object here. Missing values are never invented. + """ + + payload = self._normalize_dhcp_payload(settings) + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + current = self.dhcp(timeout=recovery_timeout) + return self._write_dhcp_payload( + payload, + current=current, + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + def static_reservation_list( + self, + *, + timeout: float | None = None, + ) -> list[StaticReservation]: + """Return normalized static DHCP reservation items.""" + + response = self.static_reservations(timeout=timeout) + dhcp = response.get("dhcp") + if not isinstance(dhcp, Mapping): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return a dhcp object" + ) + raw = dhcp.get("data") + if not isinstance(raw, list): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return dhcp.data as a list" + ) + result: list[StaticReservation] = [] + for item in raw: + if not isinstance(item, Mapping): + raise ProtocolError("static DHCP reservation is not an object") + result.append(self._normalize_reservation(item)) + return result + + def set_static_reservations( + self, + reservations: list[Mapping[str, Any]], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> list[StaticReservation]: + """Replace the complete 10-slot static DHCP reservation table.""" + + if not isinstance(reservations, list): + raise TypeError("reservations must be a list") + if len(reservations) > 10: + raise ValueError("the stock frontend supports at most 10 reservations") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + expected = [self._normalize_reservation(item) for item in reservations] + indices = [item["index"] for item in expected] + if len(indices) != len(set(indices)): + raise ValueError("reservation indices must be unique") + + macs = [item["mac"].lower() for item in expected] + if len(macs) != len(set(macs)): + raise ValueError("reservation MAC addresses must be unique") + ips = [item["ip"] for item in expected] + if len(ips) != len(set(ips)): + raise ValueError("reservation IPv4 addresses must be unique") + + address = self.address(timeout=recovery_timeout) + self._validate_reservations_against_lan(expected, address) + + current = self.static_reservation_list(timeout=recovery_timeout) + if self._reservation_cmp(current) == self._reservation_cmp(expected): + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [{ + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": expected}, + "timeout": write_timeout, + }], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: list[StaticReservation] | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.static_reservation_list(timeout=recovery_timeout) + last_actual = actual + if self._reservation_cmp(actual) == self._reservation_cmp(expected): + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected_count": len(expected), + "actual_count": len(last_actual) if last_actual is not None else None, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + raise APIError( + "static DHCP reservation write could not be verified by read-back", + method_id="router/router_set_dhcp_static_ip", + response=details, + ) + + def set_address_legacy( + self, + lan_ip: str, + lan_netmask: str, + *, + write_timeout: float = 30.0, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + force: bool = False, + ) -> LANAddressResponse: + """Use the deprecated dedicated LAN-address setter and require read-back. + + `force=True` is intended for explicit transport verification when the + caller wants to execute the setter even though the requested address + already matches the current state. + """ + + _validate_ip(lan_ip, version=4, field="lan_ip") + _validate_ip(lan_netmask, version=4, field="lan_netmask") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + if not isinstance(force, bool): + raise TypeError("force must be a bool") + + current = self.address(timeout=recovery_timeout) + router = current.get("router") + if ( + not force + and isinstance(router, Mapping) + and router.get("lan_ip") == lan_ip + and router.get("lan_netmask") == lan_netmask + ): + return current + + write_error: NR2301Error | None = None + try: + self._client.call( + "router", + "router_set_lan_ip", + data={"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + timeout=write_timeout, + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: LANAddressResponse | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.address(timeout=recovery_timeout) + last_actual = actual + router = actual.get("router") + if isinstance(router, Mapping) and router.get("lan_ip") == lan_ip and router.get("lan_netmask") == lan_netmask: + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": {"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + "actual": last_actual, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + raise APIError( + "legacy LAN address write could not be verified by read-back", + method_id="router/router_set_lan_ip", + response=details, + ) + def dns(self, *, timeout: float | None = None) -> DNSSettings: """Return the five DNS fields from the combined DHCP object.""" @@ -249,25 +484,47 @@ def _set_dns_fields( recovery_delay: float, recovery_timeout: float, ) -> DNSSettings: - if write_timeout <= 0: - raise ValueError("write_timeout must be greater than zero") - if recovery_attempts <= 0: - raise ValueError("recovery_attempts must be greater than zero") - if recovery_delay < 0: - raise ValueError("recovery_delay must not be negative") - if recovery_timeout <= 0: - raise ValueError("recovery_timeout must be greater than zero") - before = self.dhcp() - missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in before] - if missing: - raise ProtocolError( - "refusing combined DHCP write because the read-back object is " - f"missing required fields: {', '.join(missing)}" - ) - payload: dict[str, Any] = dict(before) payload.update(expected) + normalized = self._normalize_dhcp_payload(payload) + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + verified = self._write_dhcp_payload( + normalized, + current=before, + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + return cast( + DNSSettings, + { + key: verified[key] + for key in ("dnsmode", "dns1", "dns2", "ipv6dns1", "ipv6dns2") + }, + ) + + def _write_dhcp_payload( + self, + payload: dict[str, str], + *, + current: Mapping[str, Any], + write_timeout: int, + recovery_attempts: int, + recovery_delay: float, + recovery_timeout: float, + ) -> DHCPSettings: + current_cmp = { + key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if current_cmp == payload: + return cast(DHCPSettings, dict(current)) write_error: NR2301Error | None = None try: @@ -283,18 +540,18 @@ def _set_dns_fields( timeout=float(write_timeout), ) except (TransportError, ProtocolError) as exc: - # The documented write may reset management TCP. The write outcome - # is therefore determined by read-back, not by transport success. write_error = exc - last_actual: DNSSettings | None = None + last_actual: DHCPSettings | None = None last_error: NR2301Error | None = None - for attempt in range(recovery_attempts): try: - actual = self.dns(timeout=recovery_timeout) + actual = self.dhcp(timeout=recovery_timeout) last_actual = actual - if actual == expected: + actual_cmp = { + key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if actual_cmp == payload: return actual except NR2301Error as exc: last_error = exc @@ -308,8 +565,15 @@ def _set_dns_fields( time.sleep(recovery_delay) details: dict[str, Any] = { - "expected": dict(expected), - "actual": dict(last_actual) if last_actual is not None else None, + "expected": payload, + "actual": ( + { + key: last_actual.get(key) + for key in _REQUIRED_COMBINED_FIELDS + } + if last_actual is not None + else None + ), } if write_error is not None: details["write_transport_error"] = type(write_error).__name__ @@ -317,13 +581,186 @@ def _set_dns_fields( details["last_recovery_error"] = type(last_error).__name__ raise APIError( - "DNS write could not be verified by exact read-back; " - "the router state may be unchanged or the management connection " - "may still be recovering", + "combined DHCP write could not be verified by exact read-back", method_id="router/router_set_dhcp_settings_comb", response=details, ) + @classmethod + def _normalize_dhcp_payload( + cls, + settings: Mapping[str, Any], + ) -> dict[str, str]: + if not isinstance(settings, Mapping): + raise TypeError("settings must be a mapping") + + missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] + if missing: + raise ProtocolError( + "refusing combined DHCP write because the read-back object is " + "missing required fields: " + ", ".join(missing) + ) + + payload: dict[str, str] = {} + for key in _REQUIRED_COMBINED_FIELDS: + value = settings[key] + if not isinstance(value, str): + raise TypeError(f"{key} must be a str") + payload[key] = value + + cls._validate_dhcp_payload(payload) + return payload + + @staticmethod + def _validate_dhcp_payload(payload: Mapping[str, str]) -> None: + if payload["disabled"] not in {"0", "1"}: + raise ValueError("disabled must be '0' or '1'") + + for field in ("lan_ip", "start", "end"): + _validate_ip(payload[field], version=4, field=field) + + try: + ipaddress.IPv4Network( + f"0.0.0.0/{payload['lan_netmask']}", + strict=False, + ) + except ValueError as exc: + raise ValueError("lan_netmask must be a valid IPv4 netmask") from exc + + try: + lease = int(payload["leasetime"]) + except ValueError as exc: + raise ValueError("leasetime must be an integer string") from exc + if lease < 60 or lease > 604800: + raise ValueError("leasetime must be between 60 and 604800 seconds") + + try: + mtu = int(payload["mtu"]) + except ValueError as exc: + raise ValueError("mtu must be an integer string") from exc + if mtu < 1280 or mtu > 1500: + raise ValueError("mtu must be between 1280 and 1500") + + if payload["dnsmode"] not in {"auto", "manual"}: + raise ValueError("dnsmode must be 'auto' or 'manual'") + + _validate_optional_ip(payload["dns1"], version=4, field="dns1") + _validate_optional_ip(payload["dns2"], version=4, field="dns2") + _validate_optional_ip(payload["ipv6dns1"], version=6, field="ipv6dns1") + _validate_optional_ip(payload["ipv6dns2"], version=6, field="ipv6dns2") + + @staticmethod + def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: + raw_index = item.get("index") + if isinstance(raw_index, bool): + raise TypeError("reservation index must be a string or integer") + try: + index = int(raw_index) + except (TypeError, ValueError) as exc: + raise ValueError("reservation index must be numeric") from exc + if index < 0 or index > 9: + raise ValueError("reservation index must be between 0 and 9") + + mac = item.get("mac") + if not isinstance(mac, str): + raise TypeError("reservation mac must be a str") + + # ACIY.3 accepts colon-separated MACs on write but live getter + # read-back canonicalizes the same value to uppercase hyphen-separated + # form. Accept either representation and expose one stable SDK form. + separator = ":" if ":" in mac else "-" if "-" in mac else None + parts = mac.split(separator) if separator is not None else [] + if ( + len(parts) != 6 + or any( + len(part) != 2 + or any(ch not in "0123456789abcdefABCDEF" for ch in part) + for part in parts + ) + ): + raise ValueError( + "reservation mac must be a colon- or hyphen-separated MAC address" + ) + + ip = item.get("ip") + if not isinstance(ip, str): + raise TypeError("reservation ip must be a str") + _validate_ip(ip, version=4, field="reservation ip") + + normalized_mac = ":".join(part.lower() for part in parts) + first_octet = int(parts[0], 16) + if first_octet & 1: + raise ValueError("reservation mac must not be multicast") + + return StaticReservation( + index=index, + mac=normalized_mac, + ip=str(ipaddress.ip_address(ip)), + ) + + @staticmethod + def _validate_reservations_against_lan( + items: list[StaticReservation], + address: Mapping[str, Any], + ) -> None: + router = address.get("router") + if not isinstance(router, Mapping): + raise ProtocolError( + "router/router_get_lan_ip did not return a router object" + ) + lan_ip = router.get("lan_ip") + lan_netmask = router.get("lan_netmask") + if not isinstance(lan_ip, str) or not isinstance(lan_netmask, str): + raise ProtocolError( + "router/router_get_lan_ip did not return usable LAN address fields" + ) + try: + network = ipaddress.IPv4Network( + f"{lan_ip}/{lan_netmask}", + strict=False, + ) + except ValueError as exc: + raise ProtocolError( + "router/router_get_lan_ip returned an invalid LAN/network pair" + ) from exc + + for item in items: + candidate = ipaddress.IPv4Address(item["ip"]) + if candidate not in network: + raise ValueError( + f"reservation ip {item['ip']} must be inside current LAN subnet " + f"{network.with_netmask}" + ) + + @staticmethod + def _reservation_cmp( + items: list[StaticReservation], + ) -> list[tuple[int, str, str]]: + return sorted( + ( + (item["index"], item["mac"].lower(), item["ip"]) + for item in items + ), + key=lambda value: value[0], + ) + + @staticmethod + def _validate_recovery_options( + *, + write_timeout: float, + recovery_attempts: int, + recovery_delay: float, + recovery_timeout: float, + ) -> None: + if write_timeout <= 0: + raise ValueError("write_timeout must be greater than zero") + if recovery_attempts <= 0: + raise ValueError("recovery_attempts must be greater than zero") + if recovery_delay < 0: + raise ValueError("recovery_delay must not be negative") + if recovery_timeout <= 0: + raise ValueError("recovery_timeout must be greater than zero") + @staticmethod def _extract_dhcp(response: Mapping[str, Any]) -> Mapping[str, Any]: dhcp = response.get("dhcp") diff --git a/tests/integration/test_lan_router_writes.py b/tests/integration/test_lan_router_writes.py new file mode 100644 index 0000000..b46f698 --- /dev/null +++ b/tests/integration/test_lan_router_writes.py @@ -0,0 +1,281 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import ipaddress +import os + +import pytest + +from nr2301 import NR2301Client + + +if os.environ.get("NR2301_DESTRUCTIVE_INTEGRATION") != "1": + pytest.skip( + "physical LAN/router write tests require " + "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") + + 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 _dhcp_fingerprint(value): + keys = ( + "disabled", + "lan_ip", + "lan_netmask", + "start", + "end", + "leasetime", + "mtu", + "dnsmode", + "dns1", + "dns2", + "ipv6dns1", + "ipv6dns2", + ) + return tuple((key, value.get(key)) for key in keys) + + +def _reservation_fingerprint(items): + return sorted( + ( + str(item["index"]), + str(item["mac"]).lower(), + str(item["ip"]), + ) + for item in items + ) + + +def _synthetic_reservation(original, dhcp): + used_indices = {int(item["index"]) for item in original} + free = next((index for index in range(10) if index not in used_indices), None) + + lan_ip = ipaddress.IPv4Address(dhcp["lan_ip"]) + network = ipaddress.IPv4Network( + f"{dhcp['lan_ip']}/{dhcp['lan_netmask']}", + strict=False, + ) + pool_start = ipaddress.IPv4Address(dhcp["start"]) + pool_end = ipaddress.IPv4Address(dhcp["end"]) + used_ips = { + ipaddress.IPv4Address(item["ip"]) + for item in original + } + + candidate_ip = None + lower = int(network.network_address) + 1 + upper = int(network.broadcast_address) - 1 + for raw in range(upper, lower - 1, -1): + candidate = ipaddress.IPv4Address(raw) + if candidate == lan_ip: + continue + if candidate in used_ips: + continue + if pool_start <= candidate <= pool_end: + continue + candidate_ip = str(candidate) + break + + if candidate_ip is None: + raise AssertionError( + "no free static-reservation address exists inside the LAN subnet " + "and outside the DHCP pool" + ) + + used_macs = {str(item["mac"]).lower() for item in original} + mac = next( + ( + candidate + for candidate in ( + "02:00:00:00:00:fe", + "02:00:00:00:00:fd", + "02:00:00:00:00:fc", + ) + if candidate not in used_macs + ), + None, + ) + if mac is None: + raise AssertionError("could not construct a distinct synthetic MAC") + + if free is not None: + candidate = {"index": free, "mac": mac, "ip": candidate_ip} + return [dict(item) for item in original] + [candidate] + + # All slots occupied: temporarily replace the highest slot. The complete + # original table is restored in finally. + target = max(original, key=lambda item: int(item["index"])) + candidate = { + "index": int(target["index"]), + "mac": mac, + "ip": candidate_ip, + } + return [ + candidate if int(item["index"]) == int(target["index"]) else dict(item) + for item in original + ] + + +def test_lan_router_write_lifecycle_and_exact_restore(router): + original_dhcp = router.lan.dhcp(timeout=5.0) + original_reservations = router.lan.static_reservation_list(timeout=5.0) + original_address = router.lan.address(timeout=5.0) + original_mode = router.device.work_mode(timeout=5.0) + + original_dhcp_fp = _dhcp_fingerprint(original_dhcp) + original_reservation_fp = _reservation_fingerprint(original_reservations) + + router_address = original_address.get("router") + if not isinstance(router_address, dict): + pytest.fail("router_get_lan_ip returned no router object") + lan_ip = router_address.get("lan_ip") + lan_netmask = router_address.get("lan_netmask") + if not isinstance(lan_ip, str) or not isinstance(lan_netmask, str): + pytest.fail("router_get_lan_ip returned invalid address fields") + + mode = original_mode.get("work_mode") + if mode not in {"router", "bridge"}: + pytest.fail("router_get_work_mode returned an unknown mode") + + mutated_dhcp = dict(original_dhcp) + mutated_dhcp["leasetime"] = ( + "43200" if original_dhcp.get("leasetime") != "43200" else "86400" + ) + synthetic_reservations = _synthetic_reservation( + [dict(item) for item in original_reservations], + original_dhcp, + ) + + dhcp_attempted = False + reservations_attempted = False + + try: + dhcp_attempted = True + verified_dhcp = router.lan.set_dhcp_settings( + mutated_dhcp, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _dhcp_fingerprint(verified_dhcp) == _dhcp_fingerprint(mutated_dhcp) + assert _dhcp_fingerprint(verified_dhcp) != original_dhcp_fp + print( + "LAN_DHCP_WRITE" + " field=leasetime" + " changed=True" + " readback=True", + flush=True, + ) + + reservations_attempted = True + verified_reservations = router.lan.set_static_reservations( + synthetic_reservations, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _reservation_fingerprint(verified_reservations) == ( + _reservation_fingerprint(synthetic_reservations) + ) + assert _reservation_fingerprint(verified_reservations) != ( + original_reservation_fp + ) + print( + "LAN_STATIC_RESERVATION_WRITE" + f" original_count={len(original_reservations)}" + f" synthetic_count={len(synthetic_reservations)}" + " readback=True", + flush=True, + ) + + verified_address = router.lan.set_address_legacy( + lan_ip, + lan_netmask, + force=True, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + verified_router = verified_address.get("router") + assert isinstance(verified_router, dict) + assert verified_router.get("lan_ip") == lan_ip + assert verified_router.get("lan_netmask") == lan_netmask + print( + "LAN_LEGACY_ADDRESS_WRITE force_same_state=True readback=True", + flush=True, + ) + + verified_mode = router.device.set_work_mode( + mode, + force=True, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert verified_mode.get("work_mode") == mode + print( + "ROUTER_WORK_MODE_WRITE" + f" force_same_state=True" + f" mode_preserved=True", + flush=True, + ) + + finally: + if reservations_attempted: + restored_reservations = router.lan.set_static_reservations( + [dict(item) for item in original_reservations], + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _reservation_fingerprint(restored_reservations) == ( + original_reservation_fp + ) + + if dhcp_attempted: + restored_dhcp = router.lan.set_dhcp_settings( + dict(original_dhcp), + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _dhcp_fingerprint(restored_dhcp) == original_dhcp_fp + + final_dhcp = router.lan.dhcp(timeout=5.0) + final_reservations = router.lan.static_reservation_list(timeout=5.0) + final_address = router.lan.address(timeout=5.0) + final_mode = router.device.work_mode(timeout=5.0) + + assert _dhcp_fingerprint(final_dhcp) == original_dhcp_fp + assert _reservation_fingerprint(final_reservations) == original_reservation_fp + assert final_address == original_address + assert final_mode == original_mode + + print( + "LAN_ROUTER_FINAL" + " dhcp_restored=True" + " reservations_restored=True" + " address_preserved=True" + " work_mode_preserved=True", + flush=True, + ) diff --git a/tests/test_device_namespace.py b/tests/test_device_namespace.py index a64ca48..2223cab 100644 --- a/tests/test_device_namespace.py +++ b/tests/test_device_namespace.py @@ -65,7 +65,7 @@ def test_device_internet_preserves_documented_raw_access_value(): def test_work_mode_preserves_unknown_raw_mode_without_coercion(): - payload = {"mode": "future-mode", "result": 0} + payload = {"work_mode": "future-mode"} client, _ = authenticated_client(payload) assert client.device.work_mode() == payload @@ -210,3 +210,81 @@ def test_set_sleep_wait_time_raises_when_readback_does_not_match(): assert exc_info.value.method_id == "aoc/set_sleep_wait_time" assert exc_info.value.response == {"expected": 20, "actual": 30} + + + +def test_set_work_mode_uses_verified_multicall_and_readback(): + client, session = authenticated_client( + {"work_mode": "router"}, + {"responses": [{"result": 0}]}, + {"work_mode": "bridge"}, + ) + + result = client.device.set_work_mode( + "bridge", + recovery_attempts=1, + ) + + assert result == {"work_mode": "bridge"} + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + _, _, kwargs = session.calls[1] + assert kwargs["params"] == {"multicalls": 1} + assert kwargs["json"] == { + "requests": [ + { + "path": "router", + "method": "router_set_work_mode", + "data": {"work_mode": "bridge"}, + "timeout": 30, + } + ] + } + + +def test_set_work_mode_force_executes_same_state_write(): + client, session = authenticated_client( + {"work_mode": "router"}, + {"responses": [{"result": 0}]}, + {"work_mode": "router"}, + ) + + result = client.device.set_work_mode( + "router", + force=True, + recovery_attempts=1, + ) + + assert result["work_mode"] == "router" + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + + +def test_set_work_mode_uses_readback_after_transport_failure(): + client, _ = authenticated_client( + {"work_mode": "router"}, + FakeResponse({}, status_code=500), + {"work_mode": "bridge"}, + ) + + result = client.device.set_work_mode( + "bridge", + recovery_attempts=1, + ) + + assert result["work_mode"] == "bridge" + + +def test_set_work_mode_same_state_without_force_avoids_write(): + client, session = authenticated_client({"work_mode": "router"}) + + assert client.device.set_work_mode("router")["work_mode"] == "router" + assert [call[0] for call in session.calls] == ["GET"] + + +@pytest.mark.parametrize("mode", ["", "Router", "modem", "invalid"]) +def test_set_work_mode_rejects_unknown_mode_before_network(mode): + client, session = authenticated_client() + + with pytest.raises(ValueError, match="router.*bridge"): + client.device.set_work_mode(mode) + + assert session.calls == [] diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index d1d174e..0d36e57 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -226,3 +226,329 @@ def test_legacy_dhcp_settings_rejects_malformed_multicall(payload): with pytest.raises(ProtocolError): client.lan.legacy_dhcp_settings() + + + +def test_set_dhcp_settings_writes_complete_multicall_object_and_verifies(): + before = dhcp_payload() + after = dhcp_payload(leasetime="43200") + client, session = authenticated_client( + [(before, 200), ({"responses": [{"result": 0}]}, 200), (after, 200)] + ) + + requested = dict(after["dhcp"]) + result = client.lan.set_dhcp_settings(requested, recovery_attempts=1) + + assert result["leasetime"] == "43200" + method, _, kwargs = session.calls[1] + assert method == "POST" + assert kwargs["params"] == {"multicalls": 1} + member = kwargs["json"]["requests"][0] + assert member["path"] == "router" + assert member["method"] == "router_set_dhcp_settings_comb" + assert member["timeout"] == 30 + assert member["data"] == requested + assert set(member["data"]) == { + "disabled", + "lan_ip", + "lan_netmask", + "start", + "end", + "leasetime", + "mtu", + "dnsmode", + "dns1", + "dns2", + "ipv6dns1", + "ipv6dns2", + } + + +def test_set_dhcp_settings_uses_readback_after_transport_failure(): + before = dhcp_payload() + after = dhcp_payload(mtu="1499") + client, _ = authenticated_client( + [(before, 200), ({}, 500), (after, 200)] + ) + + result = client.lan.set_dhcp_settings( + dict(after["dhcp"]), + recovery_attempts=1, + ) + + assert result["mtu"] == "1499" + + +def test_set_dhcp_settings_rejects_partial_or_non_string_fields(): + incomplete = dict(dhcp_payload()["dhcp"]) + incomplete.pop("mtu") + client, session = authenticated_client([]) + + with pytest.raises(ProtocolError, match="missing required fields"): + client.lan.set_dhcp_settings(incomplete) + + wrong_type = dict(dhcp_payload()["dhcp"]) + wrong_type["mtu"] = 1500 + with pytest.raises(TypeError, match="mtu must be a str"): + client.lan.set_dhcp_settings(wrong_type) + + assert session.calls == [] + + +def test_static_reservation_list_normalizes_verified_item_shape(): + payload = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02-AA-BB-CC-DD-EE", + "ip": "192.0.2.10", + } + ], + } + } + client, _ = authenticated_client([(payload, 200)]) + + assert client.lan.static_reservation_list() == [ + { + "index": 0, + "mac": "02:aa:bb:cc:dd:ee", + "ip": "192.0.2.10", + } + ] + + +def test_set_static_reservations_uses_multicall_and_exact_readback(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + before = {"dhcp": {"cnt": 0, "data": []}} + expected_item = { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + after = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02-00-00-00-00-01", + "ip": "192.168.1.254", + } + ], + } + } + client, session = authenticated_client( + [ + (address, 200), + (before, 200), + ({"responses": [{"result": 0}]}, 200), + (after, 200), + ] + ) + + result = client.lan.set_static_reservations( + [expected_item], + recovery_attempts=1, + ) + + assert result == [expected_item] + method, _, kwargs = session.calls[2] + assert method == "POST" + assert kwargs["params"] == {"multicalls": 1} + member = kwargs["json"]["requests"][0] + assert member == { + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": [expected_item]}, + "timeout": 30, + } + + +def test_set_static_reservations_recovers_after_write_transport_failure(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + before = {"dhcp": {"cnt": 0, "data": []}} + expected_item = { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + after = {"dhcp": {"cnt": 1, "data": [expected_item]}} + client, _ = authenticated_client( + [(address, 200), (before, 200), ({}, 500), (after, 200)] + ) + + assert client.lan.set_static_reservations( + [expected_item], + recovery_attempts=1, + ) == [expected_item] + + +@pytest.mark.parametrize( + "item", + [ + {"index": "10", "mac": "02:00:00:00:00:01", "ip": "192.0.2.10"}, + {"index": "0", "mac": "invalid", "ip": "192.0.2.10"}, + {"index": "0", "mac": "02:00-AA:BB:CC:DD", "ip": "192.0.2.10"}, + {"index": "0", "mac": "02:00:00:00:00:01", "ip": "not-an-ip"}, + ], +) +def test_set_static_reservations_rejects_invalid_items_before_network(item): + client, session = authenticated_client([]) + + with pytest.raises((TypeError, ValueError)): + client.lan.set_static_reservations([item]) + + assert session.calls == [] + + +def test_set_address_legacy_force_executes_same_state_setter_and_readback(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client( + [(address, 200), ({"router": {"setting_response": "OK"}}, 200), (address, 200)] + ) + + result = client.lan.set_address_legacy( + "192.168.1.1", + "255.255.255.0", + force=True, + recovery_attempts=1, + ) + + assert result == address + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + _, _, kwargs = session.calls[1] + assert kwargs["params"]["path"] == "router" + assert kwargs["params"]["method"] == "router_set_lan_ip" + assert kwargs["json"] == { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + + +def test_set_address_legacy_same_state_without_force_avoids_write(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client([(address, 200)]) + + assert client.lan.set_address_legacy( + "192.168.1.1", + "255.255.255.0", + ) == address + + assert [call[0] for call in session.calls] == ["GET"] + + + +def test_set_static_reservations_rejects_ip_outside_current_lan_before_write(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client([(address, 200)]) + + with pytest.raises(ValueError, match="inside current LAN subnet"): + client.lan.set_static_reservations( + [ + { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.0.2.254", + } + ] + ) + + assert len(session.calls) == 1 + assert session.calls[0][0] == "GET" + + +def test_set_static_reservations_sends_numeric_index_even_if_input_is_string(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + before = {"dhcp": {"cnt": 0, "data": []}} + after = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + ], + } + } + client, session = authenticated_client( + [ + (address, 200), + (before, 200), + ({"responses": [{"data": {"dhcp": {"setting_response": "OK"}}}]}, 200), + (after, 200), + ] + ) + + client.lan.set_static_reservations( + [ + { + "index": "0", + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + ], + recovery_attempts=1, + ) + + member = session.calls[2][2]["json"]["requests"][0] + assert member["data"]["data"][0]["index"] == 0 + assert isinstance(member["data"]["data"][0]["index"], int) + + +@pytest.mark.parametrize( + "items", + [ + [ + {"index": 0, "mac": "02:00:00:00:00:01", "ip": "192.168.1.10"}, + {"index": 1, "mac": "02:00:00:00:00:01", "ip": "192.168.1.11"}, + ], + [ + {"index": 0, "mac": "02:00:00:00:00:01", "ip": "192.168.1.10"}, + {"index": 1, "mac": "02:00:00:00:00:02", "ip": "192.168.1.10"}, + ], + [ + {"index": 0, "mac": "01:00:5e:00:00:01", "ip": "192.168.1.10"}, + ], + ], +) +def test_set_static_reservations_matches_frontend_duplicate_and_multicast_checks(items): + client, session = authenticated_client([]) + + with pytest.raises(ValueError): + client.lan.set_static_reservations(items) + + assert session.calls == []