#!/usr/bin/env -S uv run --script # /// script # dependencies = ["bottle"] # /// import argparse import base64 import ipaddress import json import logging import re import signal import ssl import subprocess import threading import time import urllib.error import urllib.request import webbrowser from collections.abc import Callable from socketserver import ThreadingMixIn from wsgiref.simple_server import WSGIRequestHandler, WSGIServer, make_server import bottle logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") logger = logging.getLogger(__name__) ROUTER_HOST = "192.168.1.1" ROUTER_USERNAME = "abdus" ROUTER_PASSWORD = "xAsametk50" INTERNET_TARGET = "8.8.8.8" PUBLIC_IP_URL = "https://api.ipify.org" UNIT_SECONDS = {"day": 86400, "hour": 3600, "min": 60, "sec": 1} DNS_MODES = ("DNSProxy", "Static", "FromISP") LOG_PATH = "/cgi-bin/Log?action=GET_LOG&oid=RDM_OID_LOG_CLASSIFY&iid=[1,0,0,0,0,0]" LOG_LINE = re.compile(r"^\[(\w+)\]\s+(\w{3}\s+\d+\s+[\d:]+)\s+(\w+)\.(\w+)\s+(.*)$") PAGE = r""" Router
Router

Ping

Line

Connection

DNS for devices

Connected devices

NameIPLinkSignal
Router log 
loading...
""" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Local dashboard for the Zyxel router", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=0, help="0 picks a random free port") parser.add_argument("--timeout", type=float, default=10.0, help="Router request timeout in seconds") parser.add_argument("--no-browser", action="store_true", help="Do not open the page on start") return parser.parse_args() class Router: def __init__(self, host: str, timeout: float) -> None: self.base_url = f"https://{host}" self.timeout = timeout self.lock = threading.Lock() self.session_key = "" # Self-signed certificate; over HTTPS the router accepts a plain JSON login. context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE self.opener = urllib.request.build_opener( urllib.request.HTTPSHandler(context=context), urllib.request.HTTPCookieProcessor(), ) def request(self, path: str, body: dict | None = None, method: str | None = None) -> object: method = method or ("POST" if body is not None else "GET") separator = "&" if "?" in path else "?" suffix = f"{separator}sessionkey={self.session_key}" if self.session_key else "" headers = {"Content-Type": "application/json"} if self.session_key and method != "GET": # Since the ABSB.5 firmware, state-changing requests (POST/PUT/DELETE) are only # accepted with the session key repeated in this header; the ?sessionkey= query # param alone (still required for GETs) gets a bare 403 on everything else. headers["CSRFToken"] = self.session_key request = urllib.request.Request( url=f"{self.base_url}{path}{suffix}", data=json.dumps(body).encode() if body is not None else None, headers=headers, method=method, ) with self.opener.open(request, timeout=self.timeout) as response: raw = response.read().decode(errors="replace") return json.loads(raw) if raw.strip() else {} def login(self) -> None: self.session_key = "" reply = self.request( path="/UserLogin", body={ "Input_Account": ROUTER_USERNAME, "Input_Passwd": base64.b64encode(ROUTER_PASSWORD.encode()).decode(), "currLang": "en", "RememberPassword": 0, "SHA512_password": False, }, ) if reply.get("result") != "ZCFG_SUCCESS": raise RuntimeError(f"Login rejected: {reply}") self.session_key = reply.get("sessionkey", "") logger.info("Logged in to router") def call(self, path: str, body: dict | None = None, method: str | None = None) -> object: with self.lock: if not self.session_key: self.login() try: return self.request(path=path, body=body, method=method) except (urllib.error.URLError, json.JSONDecodeError, ConnectionError): logger.info("Router call failed, logging in again") self.login() return self.request(path=path, body=body, method=method) def reboot(self) -> object: with self.lock: if not self.session_key: self.login() try: return self.request(path="/cgi-bin/Reboot", body={}) finally: self.session_key = "" def logout(self) -> None: with self.lock: if not self.session_key: return try: self.request(path="/cgi-bin/UserLogout", body={}) logger.info("Logged out, session released") except Exception as exc: logger.warning("Logout failed, the router may hold this session until it times out: %s", exc) finally: self.session_key = "" def __enter__(self) -> "Router": self.login() return self def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: self.logout() def find_number(pattern: str, text: str) -> float | None: match = re.search(pattern, text) return float(match.group(1)) if match else None def parse_duration(text: str) -> int: return sum(int(number) * UNIT_SECONDS[unit] for number, unit in re.findall(r"(\d+)\s*(day|hour|min|sec)", text)) def parse_dsl(text: str) -> dict: state = re.search(r"Line State:\s*(\S+)", text) uptime = re.search(r"DSL uptime\s*:(.*)", text) return { "state": state.group(1) if state else None, "uptime_s": parse_duration(text=uptime.group(1)) if uptime else None, "down_kbps": find_number(pattern=r"near-end fast channel bit rate: (\d+)", text=text), "up_kbps": find_number(pattern=r"far-end fast channel bit rate: (\d+)", text=text), "snr_down_db": find_number(pattern=r"noise margin downstream: ([\d.]+)", text=text), "snr_up_db": find_number(pattern=r"noise margin upstream: ([\d.]+)", text=text), "attn_down_db": find_number(pattern=r"attenuation downstream: ([\d.]+)", text=text), "attn_up_db": find_number(pattern=r"attenuation upstream: ([\d.]+)", text=text), "crc_errors": sum(int(count) for count in re.findall(r"(?:CRC|HEC) error \w+: (\d+)", text)), "fec_errors": sum(int(count) for count in re.findall(r"FEC error \w+: (\d+)", text)), } def ping(host: str | None) -> float | None: if not host: return None result = subprocess.run( ["ping", "-c", "1", "-W", "2000", host], capture_output=True, text=True, check=False, ) return find_number(pattern=r"time=([\d.]+)", text=result.stdout) def read_device(router: Router) -> dict: info = router.call(path="/cgi-bin/DAL?oid=status")["Object"][0]["DeviceInfo"] return { "model": info.get("ModelName"), "firmware": info.get("SoftwareVersion"), "uptime_s": info.get("UpTime"), } def read_dsl(router: Router) -> dict: return parse_dsl(text=router.call(path="/cgi-bin/xDSLStatistics_handle?line=0")[0]["result"]) def read_wan(router: Router) -> dict: entries = router.call(path="/cgi-bin/DAL?oid=wan")["Object"] vdsl = next((entry for entry in entries if entry.get("Name") == "VDSL"), {}) return { "ip": vdsl.get("IPAddress") or None, "gateway": vdsl.get("GatewayIPAddress") or None, "dns": vdsl.get("DNSServer") or None, } def read_lan_dns(router: Router) -> dict: lan = router.call(path="/cgi-bin/DAL?oid=lan")["Object"][0] return { "mode": lan.get("DNS_Type"), "server1": lan.get("DNS_Servers1") or "", "server2": lan.get("DNS_Servers2") or "", } def validate_dns_server(value: str) -> str: try: address = ipaddress.IPv4Address(value.strip()) except ValueError: raise ValueError(f"{value!r} is not a valid IPv4 address") from None if address.is_unspecified or address.is_loopback or address.is_multicast or address.is_reserved: raise ValueError(f"{address} cannot be used as a DNS server") return str(address) def set_lan_dns(router: Router, mode: str, server1: str, server2: str) -> dict: if mode not in DNS_MODES: raise ValueError(f"Unknown DNS mode {mode!r}") servers = ["", ""] if mode == "Static": servers[0] = validate_dns_server(value=server1 or "") if server2: servers[1] = validate_dns_server(value=server2) if servers[0] == servers[1]: raise ValueError("Primary and secondary DNS must differ") lan = router.call(path="/cgi-bin/DAL?oid=lan")["Object"][0] lan["DNS_Type"] = mode lan["DNS_Servers1"], lan["DNS_Servers2"] = servers if mode == "Static" and not servers[1]: del lan["DNS_Servers2"] reply = router.call(path="/cgi-bin/DAL?oid=lan", body=lan, method="PUT") if reply.get("result") != "ZCFG_SUCCESS": raise RuntimeError(f"Router rejected the change: {reply.get('ReplyMsg') or reply}") logger.warning("LAN DNS set to %s %s %s", mode, servers[0], servers[1]) return read_lan_dns(router=router) def read_log(router: Router) -> dict: enabled = bool(router.call(path="/cgi-bin/DAL?oid=logset")["Object"][0]["Enable"]) reply = router.call(path=LOG_PATH) entries = [] for raw in reversed(reply[0].get("result") or []): match = LOG_LINE.match(raw.strip()) if match: category, time_text, facility, level, message = match.groups() entries.append({"time": time_text, "category": category, "level": level, "message": message}) else: entries.append({"time": "", "category": "", "level": "", "message": raw}) return {"enabled": enabled, "entries": entries} def set_logging(router: Router, enabled: bool) -> dict: logset = router.call(path="/cgi-bin/DAL?oid=logset")["Object"][0] logset["Enable"] = enabled reply = router.call(path="/cgi-bin/DAL?oid=logset", body=logset, method="PUT") if reply.get("result") != "ZCFG_SUCCESS": raise RuntimeError(f"Router rejected the change: {reply.get('ReplyMsg') or reply}") logger.warning("Router logging %s", "enabled" if enabled else "disabled") return read_log(router=router) def fetch_public_ip(timeout: float) -> str | None: try: with urllib.request.urlopen(PUBLIC_IP_URL, timeout=timeout) as response: return response.read().decode().strip() except Exception as exc: logger.info("Could not fetch public IP: %s", exc) return None def find_active_pppoe(status: dict) -> dict: candidates = [ entry for entry in status["Object"][0]["WanLanInfo"] if entry.get("Encapsulation") == "PPPoE" and entry.get("Status") == "Up" ] if not candidates: raise RuntimeError("No active PPPoE WAN session found") return candidates[0] def wan_ip_of(entry: dict) -> str | None: return next(iter(entry.get("IPv4Address") or []), {}).get("IPAddress") def wan_action(router: Router, wanppp_idx: int, action: str) -> None: # Disconnect/Connect tear down or rebuild the PPPoE link, so the router often doesn't answer in # time (or at all, if the link it was answering over just dropped). That doesn't mean the action # failed: treat it as "sent, outcome unknown" and let the caller confirm by polling the real state. try: reply = router.call(path="/cgi-bin/DAL?oid=status", body={"action": action, "wanpppIdx": wanppp_idx}, method="PUT") except Exception as exc: logger.info("%s request did not get a response (%s), checking the actual state instead", action, exc) return if reply.get("result") not in (None, "ZCFG_SUCCESS"): raise RuntimeError(f"Router rejected {action}: {reply}") def find_wan_entry(status: dict, wanppp_idx: int) -> dict | None: return next( (e for e in status["Object"][0]["WanLanInfo"] if e.get("wanpppIdx") == wanppp_idx and e.get("Encapsulation") == "PPPoE"), None, ) def poll_until_up(router: Router, wanppp_idx: int, timeout: float) -> dict | None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: entry = find_wan_entry(status=router.call(path="/cgi-bin/DAL?oid=status"), wanppp_idx=wanppp_idx) if entry and entry.get("Status") == "Up" and wan_ip_of(entry=entry): return entry except Exception as exc: logger.info("Still reconnecting: %s", exc) time.sleep(2.0) return None def refresh_public_ip(router: Router, wait_after_disconnect: float = 5.0, max_wait: float = 60.0, connect_retries: int = 2) -> dict: """Drop and re-establish the PPPoE session, so the ISP's CGNAT pool may hand out a different public IP. The router's own auto-reconnect doesn't always kick in on its own (it can get stuck at N/A), so a Connect is retried if the link isn't back up yet instead of leaving it down. """ wanppp_idx = find_active_pppoe(status=router.call(path="/cgi-bin/DAL?oid=status"))["wanpppIdx"] wan_action(router=router, wanppp_idx=wanppp_idx, action="Disconnect") time.sleep(wait_after_disconnect) wan_action(router=router, wanppp_idx=wanppp_idx, action="Connect") per_attempt = max(max_wait / (connect_retries + 1), 15.0) for attempt in range(connect_retries + 1): entry = poll_until_up(router=router, wanppp_idx=wanppp_idx, timeout=per_attempt) if entry: logger.warning("PPPoE session refreshed, new WAN IP %s", wan_ip_of(entry=entry)) return {"wan_ip": wan_ip_of(entry=entry), "public_ip": fetch_public_ip(timeout=10)} if attempt < connect_retries: logger.warning("Not back up yet, retrying Connect (%d/%d)...", attempt + 1, connect_retries) wan_action(router=router, wanppp_idx=wanppp_idx, action="Connect") raise TimeoutError(f"WAN session did not come back up within {max_wait:.0f}s, even after retrying Connect") def read_hosts(router: Router) -> list[dict]: hosts = router.call(path="/cgi-bin/DAL?oid=lanhosts")["Object"][0]["lanhosts"] return [ { "name": host.get("HostName") or host.get("DeviceName") or "unknown", "ip": host.get("IPAddress"), "mac": host.get("PhysAddress"), "connection": host.get("X_ZYXEL_ConnectionType"), "active": bool(host.get("Active")), "signal": host.get("X_ZYXEL_SignalStrength"), } for host in hosts ] class Poller: """Refreshes each value on its own interval in a daemon thread, so requests never wait on the slow router.""" def __init__(self, router: Router) -> None: self.state: dict[str, object] = {} self.tasks: dict[str, tuple[Callable[[], object], float]] = { "router_ms": (lambda: ping(host=ROUTER_HOST), 3), "internet_ms": (lambda: ping(host=INTERNET_TARGET), 3), "dsl": (lambda: read_dsl(router=router), 10), "device": (lambda: read_device(router=router), 15), "wan": (lambda: read_wan(router=router), 15), "isp_gateway_ms": (self.ping_gateway, 15), "hosts": (lambda: read_hosts(router=router), 30), "dns": (lambda: read_lan_dns(router=router), 30), "public_ip": (lambda: fetch_public_ip(timeout=10), 120), } def ping_gateway(self) -> float | None: wan = self.state.get("wan") return ping(host=wan["gateway"] if wan and wan["ip"] else None) def loop(self, name: str, task: Callable[[], object], interval: float) -> None: while True: started = time.monotonic() try: self.state[name] = task() except Exception as exc: logger.warning("%s failed: %s", name, exc) self.state[name] = None time.sleep(max(0.5, interval - (time.monotonic() - started))) def start(self) -> None: for name, (task, interval) in self.tasks.items(): threading.Thread(target=self.loop, args=(name, task, interval), daemon=True).start() def overview(self) -> dict: state = dict(self.state) return { "device": state.get("device"), "dsl": state.get("dsl"), "wan": state.get("wan"), "hosts": state.get("hosts"), "dns": state.get("dns"), "public_ip": state.get("public_ip"), "ping": { "router_ms": state.get("router_ms"), "isp_gateway_ms": state.get("isp_gateway_ms"), "internet_ms": state.get("internet_ms"), }, } def build_app(router: Router, poller: Poller) -> bottle.Bottle: app = bottle.Bottle() @app.get("/") def index() -> str: return PAGE @app.get("/api/overview") def api_overview() -> dict: return poller.overview() @app.get("/api/log") def api_log() -> str: bottle.response.content_type = "application/json" try: return json.dumps(read_log(router=router)) except Exception as exc: logger.warning("log failed: %s", exc) bottle.response.status = 502 return json.dumps({"error": str(exc)}) @app.post("/api/log") def api_log_toggle() -> str: bottle.response.content_type = "application/json" enabled = (bottle.request.json or {}).get("enabled") if not isinstance(enabled, bool): bottle.response.status = 400 return json.dumps({"error": "enabled must be true or false"}) try: return json.dumps(set_logging(router=router, enabled=enabled)) except Exception as exc: logger.warning("logging toggle failed: %s", exc) bottle.response.status = 502 return json.dumps({"error": str(exc)}) @app.post("/api/dns") def api_dns() -> str: bottle.response.content_type = "application/json" payload = bottle.request.json or {} try: result = set_lan_dns( router=router, mode=payload.get("mode", ""), server1=payload.get("server1", ""), server2=payload.get("server2", ""), ) except ValueError as exc: bottle.response.status = 400 return json.dumps({"error": str(exc)}) except Exception as exc: logger.warning("DNS change failed: %s", exc) bottle.response.status = 502 return json.dumps({"error": str(exc)}) poller.state["dns"] = result return json.dumps(result) @app.post("/api/new-ip") def api_new_ip() -> str: bottle.response.content_type = "application/json" try: result = refresh_public_ip(router=router) except Exception as exc: logger.warning("Public IP refresh failed: %s", exc) bottle.response.status = 502 return json.dumps({"error": str(exc)}) poller.state["public_ip"] = result["public_ip"] return json.dumps(result) @app.post("/api/reboot") def api_reboot() -> dict: logger.warning("Rebooting router on request") try: router.reboot() except Exception as exc: # The router may drop the connection while it goes down. logger.info("Reboot call ended with: %s", exc) return {"ok": True} return app class ThreadingServer(ThreadingMixIn, WSGIServer): daemon_threads = True class QuietHandler(WSGIRequestHandler): def log_message(self, format: str, *args: object) -> None: pass def raise_system_exit(signum: int, frame: object) -> None: raise SystemExit(0) def main() -> None: args = parse_args() # pkill sends SIGTERM by default; without this it kills the process before the `with` block # below can log out, leaking the router session until it times out on its own. signal.signal(signal.SIGTERM, raise_system_exit) with Router(host=ROUTER_HOST, timeout=args.timeout) as router: poller = Poller(router=router) poller.start() server = make_server( host=args.host, port=args.port, app=build_app(router=router, poller=poller), server_class=ThreadingServer, handler_class=QuietHandler, ) url = f"http://{args.host}:{server.server_port}" logger.info("Serving %s", url) if not args.no_browser: opener = threading.Timer(interval=0.5, function=webbrowser.open, args=[url]) opener.daemon = True opener.start() try: server.serve_forever(poll_interval=0.2) except (KeyboardInterrupt, SystemExit): logger.info("Stopping") finally: server.server_close() if __name__ == "__main__": main()