From 545098eceb1e9f6cf41512a0e3d285c8f2065877 Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Sun, 18 Oct 2026 22:50:39 +0300 Subject: [PATCH] feat(router): Add local dashboard with auto-login, status, devices and restart --- router_ui.py | 547 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 547 insertions(+) create mode 100755 router_ui.py diff --git a/router_ui.py b/router_ui.py new file mode 100755 index 0000000..f6e976d --- /dev/null +++ b/router_ui.py @@ -0,0 +1,547 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["bottle"] +# /// + +import argparse +import base64 +import json +import logging +import re +import ssl +import subprocess +import threading +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" +UNIT_SECONDS = {"day": 86400, "hour": 3600, "min": 60, "sec": 1} + +PAGE = r""" + + + + +Router + + + +
+ + +
+
Router
+ +
+ + + +

Ping

+
+ +
+ +

Line

+ +

Connection

+ +

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) -> object: + separator = "&" if "?" in path else "?" + suffix = f"{separator}sessionkey={self.session_key}" if self.session_key else "" + request = urllib.request.Request( + url=f"{self.base_url}{path}{suffix}", + data=json.dumps(body).encode() if body is not None else None, + headers={"Content-Type": "application/json"}, + method="POST" if body is not None else "GET", + ) + 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) -> object: + with self.lock: + if not self.session_key: + self.login() + try: + return self.request(path=path, body=body) + except (urllib.error.URLError, json.JSONDecodeError, ConnectionError): + logger.info("Router call failed, logging in again") + self.login() + return self.request(path=path, body=body) + + 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 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), + "errors": sum(int(count) for count in re.findall(r"(?:CRC|FEC|HEC) 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_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 + ] + + +def safely(reader, router: Router) -> object: + try: + return reader(router) + except Exception as exc: + logger.warning("%s failed: %s", reader.__name__, exc) + return None + + +def run_parallel(tasks: dict[str, Callable[[], object]]) -> dict[str, object]: + # Daemon threads, so Ctrl+C does not wait for slow router calls or pings. + results: dict[str, object] = {} + + def worker(name: str, task: Callable[[], object]) -> None: + results[name] = task() + + threads = [threading.Thread(target=worker, args=(name, task), daemon=True) for name, task in tasks.items()] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + return results + + +def overview(router: Router) -> dict: + results = run_parallel( + tasks={ + "device": lambda: safely(read_device, router), + "dsl": lambda: safely(read_dsl, router), + "wan": lambda: safely(read_wan, router), + "hosts": lambda: safely(read_hosts, router), + "router_ms": lambda: ping(host=ROUTER_HOST), + "internet_ms": lambda: ping(host=INTERNET_TARGET), + } + ) + wan = results["wan"] + gateway = wan["gateway"] if wan and wan["ip"] else None + return { + "device": results["device"], + "dsl": results["dsl"], + "wan": wan, + "hosts": results["hosts"], + "ping": { + "router_ms": results["router_ms"], + "isp_gateway_ms": ping(host=gateway), + "internet_ms": results["internet_ms"], + }, + } + + +def build_app(router: Router) -> bottle.Bottle: + app = bottle.Bottle() + + @app.get("/") + def index() -> str: + return PAGE + + @app.get("/api/overview") + def api_overview() -> dict: + return overview(router=router) + + @app.get("/api/log") + def api_log() -> dict: + bottle.response.content_type = "application/json" + try: + return json.dumps(router.call(path="/cgi-bin/DAL?oid=zlog")["Object"]) + except Exception as exc: + logger.warning("log failed: %s", exc) + bottle.response.status = 502 + return json.dumps([]) + + @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 main() -> None: + args = parse_args() + router = Router(host=ROUTER_HOST, timeout=args.timeout) + server = make_server( + host=args.host, + port=args.port, + app=build_app(router=router), + 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: + logger.info("Stopping") + finally: + server.server_close() + + +if __name__ == "__main__": + main()