#!/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
"""
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()