From d43b375d7392c8f40e0fab290bbb29ec8217b661 Mon Sep 17 00:00:00 2001 From: Ignace Date: Sat, 22 Aug 2026 19:54:38 +0200 Subject: [PATCH] bug in zyxel --- README.md | 8 +++++ batch/test_api1.py | 36 +++++++++++++++++++ batch/test_connect.py | 13 +++++++ batch/test_connect2.py | 13 +++++++ lib/_zyxel.py | 79 ++++++++++++++++++++++++++++++++++++++---- tests/test_service.py | 24 ++++++++++++- 6 files changed, 166 insertions(+), 7 deletions(-) create mode 100644 batch/test_api1.py create mode 100644 batch/test_connect.py create mode 100644 batch/test_connect2.py diff --git a/README.md b/README.md index c3c5c48..a740650 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,14 @@ logs into the router's encrypted web API and polls its active LAN clients. 3. Test connectivity: `python service.py --once` 4. Run continuously: `python service.py` +Check any MAC address directly through the Zyxel client with: + +```bash +python lib/_zyxel.py --mac AA:BB:CC:DD:EE:FF +``` + +Omit `--mac` to check `arrival_detection.device.mac` from `service.yaml`. + An arrival is logged after the phone was confirmed absent and then present. It also calls `send_notification()` from `notify.py`; its message, ntfy topic, and timeout are configured under `arrival_detection.notification`. Set diff --git a/batch/test_api1.py b/batch/test_api1.py new file mode 100644 index 0000000..7c5b918 --- /dev/null +++ b/batch/test_api1.py @@ -0,0 +1,36 @@ +import requests +import re +from urllib.parse import urljoin + +ROUTER = "https://192.168.178.2" + +r = requests.get(ROUTER, verify=False, timeout=5) + +scripts = re.findall(r']+src=["\']([^"\']+)', r.text) + +for script in scripts: + if script.endswith("/app.js"): + url = urljoin(ROUTER, script) + + print("Downloading:", url) + + js = requests.get(url, verify=False, timeout=10).text + + print("Size:", len(js), "bytes") + + # Print strings containing things that are likely to be API endpoints + print("\nInteresting URLs/strings:\n") + + patterns = [ + r'["\']([^"\']*(?:api|cgi|device|client|status|dhcp|wifi|wireless|lan)[^"\']*)["\']', + ] + + found = set() + + for pattern in patterns: + for match in re.findall(pattern, js, re.IGNORECASE): + if len(match) < 300: + found.add(match) + + for item in sorted(found): + print(item) \ No newline at end of file diff --git a/batch/test_connect.py b/batch/test_connect.py new file mode 100644 index 0000000..f5aad53 --- /dev/null +++ b/batch/test_connect.py @@ -0,0 +1,13 @@ +import requests + +ROUTER = "https://192.168.178.2" + +r = requests.get( + ROUTER, + verify=False, + timeout=5 +) + +print(r.status_code) +print(r.url) +print(r.text[:1000]) \ No newline at end of file diff --git a/batch/test_connect2.py b/batch/test_connect2.py new file mode 100644 index 0000000..3638882 --- /dev/null +++ b/batch/test_connect2.py @@ -0,0 +1,13 @@ +import requests +import re + +ROUTER = "https://192.168.178.2" + +r = requests.get(ROUTER, verify=False, timeout=5) + +# Find JavaScript files +scripts = re.findall(r']+src=([^ >]+)', r.text) + +print("JavaScript files:") +for script in scripts: + print(script) \ No newline at end of file diff --git a/lib/_zyxel.py b/lib/_zyxel.py index 317cad4..fc350fa 100644 --- a/lib/_zyxel.py +++ b/lib/_zyxel.py @@ -2,14 +2,18 @@ from __future__ import annotations +import argparse import base64 import json import os +import sys from dataclasses import dataclass +from pathlib import Path from typing import Any from urllib.parse import urlsplit import requests +import yaml from Crypto.Cipher import AES, PKCS1_v1_5 from Crypto.PublicKey import RSA from Crypto.Util.Padding import pad, unpad @@ -145,13 +149,24 @@ class ZyxelRouter: result = self.dal_get(self.hosts_oid) if result.get("result") not in (None, "ZCFG_SUCCESS"): raise ZyxelError(f"lanhosts query failed: {result.get('result')}") - objects = result.get("Object", []) - if isinstance(objects, dict): - objects = [objects] + + def host_records(value: Any): + """Yield host dictionaries from flat and nested DAL response shapes.""" + if isinstance(value, dict): + if any( + key in value + for key in ("PhysAddress", "physAddress", "MACAddr", "MacAddress") + ): + yield value + return + for child in value.values(): + yield from host_records(child) + elif isinstance(value, list): + for child in value: + yield from host_records(child) + hosts: list[LanHost] = [] - for item in objects: - if not isinstance(item, dict): - continue + for item in host_records(result.get("Object", [])): raw_mac = next((item.get(key) for key in ("PhysAddress", "physAddress", "MACAddr", "MacAddress") if item.get(key)), None) if not raw_mac: continue @@ -182,3 +197,55 @@ class ZyxelRouter: self._session_key = None self._aes_key = None self.session.close() + + +def _load_cli_config(path: Path) -> tuple[dict[str, Any], str]: + try: + with path.open(encoding="utf-8") as stream: + root = yaml.safe_load(stream) + arrival = root["arrival_detection"] + return arrival["zyxel"], str(arrival["device"]["mac"]) + except (OSError, TypeError, KeyError, yaml.YAMLError) as error: + raise ZyxelError(f"could not load configuration from {path}: {error}") from error + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Check whether a MAC address is connected to the configured Zyxel router." + ) + parser.add_argument( + "--config", + type=Path, + default=Path(__file__).resolve().parents[1] / "service.yaml", + help="service configuration file (default: project service.yaml)", + ) + parser.add_argument( + "--mac", + help="MAC address to check (default: arrival_detection.device.mac)", + ) + args = parser.parse_args() + + try: + config, configured_mac = _load_cli_config(args.config) + mac = normalize_mac(args.mac or configured_mac) + router = ZyxelRouter( + config["host"], + config["username"], + config["password"], + timeout=float(config.get("timeout", 10)), + verify_tls=bool(config.get("verify_tls", True)), + hosts_oid=str(config.get("hosts_oid", "lanhosts")), + ) + try: + connected = router.is_connected(mac) + finally: + router.close() + print(f"{mac} {'connected' if connected else 'absent'}") + return 0 + except (ValueError, ZyxelError) as error: + print(f"error: {error}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_service.py b/tests/test_service.py index 9d7ec23..89df4ac 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -3,7 +3,7 @@ from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import patch -from lib._zyxel import normalize_mac +from lib._zyxel import ZyxelRouter, normalize_mac from lib._arrival_detection import ArrivalDetector, _notify @@ -35,6 +35,28 @@ class MacTests(unittest.TestCase): with self.assertRaises(ValueError): normalize_mac("not-a-mac") + def test_parses_ex5601_nested_lanhosts_response(self): + router = ZyxelRouter("192.0.2.1", "user", "password") + router.dal_get = lambda _oid: { + "result": "ZCFG_SUCCESS", + "Object": [ + { + "lanhosts": [ + { + "PhysAddress": "2C:DB:07:50:2B:5C", + "Active": True, + "HostName": "workstation", + "IPAddress": "192.168.178.10", + } + ] + } + ], + } + hosts = router.get_lan_hosts() + self.assertEqual(len(hosts), 1) + self.assertEqual(hosts[0].mac, "2c:db:07:50:2b:5c") + self.assertTrue(router.is_connected("2c-db-07-50-2b-5c")) + class NotificationTests(unittest.TestCase): @patch("lib._arrival_detection.send_notification")