diff --git a/router_monitor.py b/router_monitor.py new file mode 100755 index 0000000..48e8bab --- /dev/null +++ b/router_monitor.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 + +import argparse +import base64 +import csv +import json +import logging +import re +import ssl +import subprocess +import time +import urllib.request +from datetime import datetime +from pathlib import Path + +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" + +COLUMNS = [ + "timestamp", + "router_uptime_s", + "dsl_state", + "dsl_uptime_s", + "down_kbps", + "up_kbps", + "snr_down_db", + "snr_up_db", + "attn_down_db", + "attn_up_db", + "dsl_errors", + "wan_ip", + "ping_router_ms", + "ping_isp_gateway_ms", + "ping_internet_ms", +] +UNIT_SECONDS = {"day": 86400, "hour": 3600, "min": 60, "sec": 1} +SLOW_PING_MS = 150 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Log DSL, PPPoE and ping health of the Zyxel router to a CSV to tell ISP from device drops", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("--output", type=Path, default=Path.cwd() / "router_monitor.csv") + parser.add_argument("--interval", type=float, default=30.0, help="Seconds between samples") + parser.add_argument("--internet-target", default="8.8.8.8") + parser.add_argument("--timeout", type=float, default=10.0, help="Router request timeout in seconds") + return parser.parse_args() + + +def build_opener() -> urllib.request.OpenerDirector: + # Self-signed certificate; over HTTPS the router accepts plain JSON login. + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + return urllib.request.build_opener( + urllib.request.HTTPSHandler(context=context), + urllib.request.HTTPCookieProcessor(), + ) + + +def login(opener: urllib.request.OpenerDirector, base_url: str, timeout: float) -> str: + request = urllib.request.Request( + url=f"{base_url}/UserLogin", + data=json.dumps( + { + "Input_Account": ROUTER_USERNAME, + "Input_Passwd": base64.b64encode(ROUTER_PASSWORD.encode()).decode(), + "currLang": "en", + "RememberPassword": 0, + "SHA512_password": False, + } + ).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with opener.open(request, timeout=timeout) as response: + body = json.loads(response.read().decode()) + if body.get("result") != "ZCFG_SUCCESS": + raise RuntimeError(f"Login rejected: {body}") + return body.get("sessionkey", "") + + +def get_json(opener: urllib.request.OpenerDirector, url: str, timeout: float) -> object: + with opener.open(url, timeout=timeout) as response: + return json.loads(response.read().decode(errors="replace")) + + +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 find_number(pattern: str, text: str) -> float | None: + match = re.search(pattern, text) + return float(match.group(1)) if match else None + + +def parse_dsl(text: str) -> dict: + state = re.search(r"Line State:\s*(\S+)", text) + uptime = re.search(r"DSL uptime\s*:(.*)", text) + errors = sum(int(count) for count in re.findall(r"(?:CRC|FEC|HEC) error \w+: (\d+)", text)) + return { + "dsl_state": state.group(1) if state else "", + "dsl_uptime_s": parse_duration(text=uptime.group(1)) if uptime else "", + "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), + "dsl_errors": errors, + } + + +def read_router(opener: urllib.request.OpenerDirector, base_url: str, session_key: str, timeout: float) -> dict: + def fetch(path: str) -> object: + return get_json(opener=opener, url=f"{base_url}{path}&sessionkey={session_key}", timeout=timeout) + + dsl = fetch("/cgi-bin/xDSLStatistics_handle?line=0") + status = fetch("/cgi-bin/DAL?oid=status") + wan = fetch("/cgi-bin/DAL?oid=wan") + + sample = parse_dsl(text=dsl[0]["result"]) + sample["router_uptime_s"] = status["Object"][0]["DeviceInfo"]["UpTime"] + vdsl = next((entry for entry in wan["Object"] if entry.get("Name") == "VDSL"), {}) + sample["wan_ip"] = vdsl.get("IPAddress", "") + sample["isp_gateway"] = vdsl.get("GatewayIPAddress", "") + return sample + + +def ping(host: str) -> 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 is_slow(sample: dict) -> bool: + return (sample.get("ping_internet_ms") or 0) > SLOW_PING_MS + + +def describe(sample: dict) -> str: + def ms(key: str) -> str: + value = sample.get(key) + return "LOST" if value is None else f"{value:.0f}ms" + + return ( + f"router {ms('ping_router_ms')} | internet {ms('ping_internet_ms')} | " + f"dsl {sample.get('dsl_state') or '?'} up {sample.get('dsl_uptime_s', '?')}s | " + f"errors {sample.get('dsl_errors', '?')} | wan {sample.get('wan_ip') or '-'}" + ) + + +def log_changes(previous: dict, current: dict) -> None: + if not previous: + return + if current.get("router_uptime_s") != "" and previous.get("router_uptime_s") != "": + if current["router_uptime_s"] < previous["router_uptime_s"]: + logger.warning("ROUTER REBOOTED (uptime reset)") + if current.get("dsl_uptime_s") != "" and previous.get("dsl_uptime_s") != "": + if current["dsl_uptime_s"] < previous["dsl_uptime_s"]: + logger.warning("DSL LINE RESYNCED (line side / ISP)") + if current.get("dsl_state") != previous.get("dsl_state"): + logger.warning("DSL state %s -> %s", previous.get("dsl_state"), current.get("dsl_state")) + if current.get("wan_ip") != previous.get("wan_ip"): + logger.warning("WAN IP %s -> %s (PPPoE session changed)", previous.get("wan_ip"), current.get("wan_ip")) + for key, label in (("ping_router_ms", "router"), ("ping_internet_ms", "internet")): + if (previous.get(key) is None) != (current.get(key) is None): + logger.warning("Ping to %s %s", label, "LOST" if current.get(key) is None else "RESTORED") + if is_slow(sample=current) != is_slow(sample=previous): + if is_slow(sample=current): + logger.warning("Internet ping SLOW: %.0f ms", current["ping_internet_ms"]) + else: + logger.warning("Internet ping back to normal") + + +def main() -> None: + args = parse_args() + base_url = f"https://{ROUTER_HOST}" + opener = build_opener() + session_key = "" + previous: dict = {} + new_file = not args.output.exists() + + with args.output.open(mode="a", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=COLUMNS, extrasaction="ignore") + if new_file: + writer.writeheader() + logger.info("Logging to %s every %.0fs", args.output, args.interval) + + while True: + started = time.monotonic() + sample = {"timestamp": datetime.now().isoformat(timespec="seconds")} + try: + if not session_key: + session_key = login(opener=opener, base_url=base_url, timeout=args.timeout) + sample.update( + read_router(opener=opener, base_url=base_url, session_key=session_key, timeout=args.timeout) + ) + except Exception as exc: + logger.warning("Router query failed: %s", exc) + session_key = "" + + sample["ping_router_ms"] = ping(host=ROUTER_HOST) + sample["ping_isp_gateway_ms"] = ping(host=sample.get("isp_gateway", "")) + sample["ping_internet_ms"] = ping(host=args.internet_target) + + log_changes(previous=previous, current=sample) + writer.writerow(sample) + handle.flush() + logger.info(describe(sample=sample)) + previous = sample + time.sleep(max(0.0, args.interval - (time.monotonic() - started))) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + logger.info("Stopped")