"""Small client for the encrypted Zyxel OPAL/DAL web API.""" from __future__ import annotations import base64 import json import os from dataclasses import dataclass from typing import Any from urllib.parse import urlsplit import requests from Crypto.Cipher import AES, PKCS1_v1_5 from Crypto.PublicKey import RSA from Crypto.Util.Padding import pad, unpad class ZyxelError(RuntimeError): """The router rejected a request or returned an unexpected response.""" def normalize_mac(value: str) -> str: compact = "".join(character for character in value if character.isalnum()).lower() if len(compact) != 12 or any(character not in "0123456789abcdef" for character in compact): raise ValueError(f"invalid MAC address: {value!r}") return ":".join(compact[index : index + 2] for index in range(0, 12, 2)) @dataclass(frozen=True) class LanHost: mac: str active: bool hostname: str = "" ip_address: str = "" class ZyxelRouter: """Read-only client for Zyxel routers using the OPAL DAL API.""" def __init__( self, host: str, username: str, password: str, *, timeout: float = 10, verify_tls: bool = True, hosts_oid: str = "lanhosts", ) -> None: if "://" not in host: host = f"http://{host}" parsed = urlsplit(host) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("zyxel.host must be a hostname/IP, or an http(s) URL") self.url = host.rstrip("/") self.username = username self.password = password self.timeout = timeout self.verify_tls = verify_tls self.hosts_oid = hosts_oid self.session = requests.Session() self._aes_key: bytes | None = None self._session_key: str | None = None def __enter__(self) -> "ZyxelRouter": self.login() return self def __exit__(self, *_: object) -> None: self.close() def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response: kwargs.setdefault("timeout", self.timeout) kwargs.setdefault("verify", self.verify_tls) try: response = self.session.request(method, f"{self.url}{path}", **kwargs) response.raise_for_status() return response except requests.RequestException as error: raise ZyxelError(f"Zyxel request failed: {error}") from error @staticmethod def _json(response: requests.Response) -> dict[str, Any]: try: value = response.json() except (requests.JSONDecodeError, ValueError) as error: raise ZyxelError("Zyxel returned a non-JSON response") from error if not isinstance(value, dict): raise ZyxelError("Zyxel returned an unexpected JSON response") return value def login(self) -> None: self._request("GET", "/GetInfoNoLogin") public_key = self._json(self._request("GET", "/getRSAPublickKey")).get("RSAPublicKey") if not public_key: raise ZyxelError("Zyxel did not return an RSA public key") self._aes_key = os.urandom(32) iv = os.urandom(32) login = { "Input_Account": self.username, "Input_Passwd": base64.b64encode(self.password.encode()).decode(), "currLang": "en", "RememberPassword": 0, } cipher = AES.new(self._aes_key, AES.MODE_CBC, iv[:16]) content = cipher.encrypt(pad(json.dumps(login, separators=(",", ":")).encode(), 16)) rsa = PKCS1_v1_5.new(RSA.import_key(public_key.encode())) encrypted_key = rsa.encrypt(base64.b64encode(self._aes_key)) payload = { "content": base64.b64encode(content).decode(), "key": base64.b64encode(encrypted_key).decode(), "iv": base64.b64encode(iv).decode(), } result = self._decrypt(self._json(self._request("POST", "/UserLogin", json=payload))) if result.get("result") != "ZCFG_SUCCESS" or not result.get("sessionkey"): raise ZyxelError(f"Zyxel login failed: {result.get('result', 'unknown error')}") self._session_key = str(result["sessionkey"]) def _decrypt(self, envelope: dict[str, Any]) -> dict[str, Any]: if "content" not in envelope or "iv" not in envelope: raise ZyxelError(str(envelope.get("result") or "unencrypted Zyxel response")) if self._aes_key is None: raise ZyxelError("not logged in") try: iv = base64.b64decode(envelope["iv"])[:16] encrypted = base64.b64decode(envelope["content"]) clear = unpad(AES.new(self._aes_key, AES.MODE_CBC, iv).decrypt(encrypted), 16) result = json.loads(clear) except (ValueError, KeyError, json.JSONDecodeError) as error: raise ZyxelError("could not decrypt Zyxel response") from error if not isinstance(result, dict): raise ZyxelError("Zyxel returned an unexpected encrypted response") return result def dal_get(self, oid: str) -> dict[str, Any]: if not self._session_key: self.login() response = self._request( "GET", "/cgi-bin/DAL", params={"oid": oid, "sessionkey": self._session_key} ) return self._decrypt(self._json(response)) def get_lan_hosts(self) -> list[LanHost]: 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] hosts: list[LanHost] = [] for item in objects: if not isinstance(item, dict): continue raw_mac = next((item.get(key) for key in ("PhysAddress", "physAddress", "MACAddr", "MacAddress") if item.get(key)), None) if not raw_mac: continue try: mac = normalize_mac(str(raw_mac)) except ValueError: continue raw_active = next((item.get(key) for key in ("Active", "active", "Enable", "enable") if key in item), True) active = raw_active if isinstance(raw_active, bool) else str(raw_active).lower() in {"1", "true", "yes", "active", "enabled"} hosts.append(LanHost( mac=mac, active=active, hostname=str(item.get("HostName") or item.get("hostName") or ""), ip_address=str(item.get("IPAddress") or item.get("ipAddress") or item.get("IPAddr") or ""), )) return hosts def is_connected(self, mac: str) -> bool: wanted = normalize_mac(mac) return any(host.mac == wanted and host.active for host in self.get_lan_hosts()) def close(self) -> None: if self._session_key: try: self._request("GET", "/cgi-bin/UserLogout", params={"sessionkey": self._session_key}) except ZyxelError: pass self._session_key = None self._aes_key = None self.session.close()