#!/usr/bin/env python3 """Apply isolated test hotspot on hAP lite. Does not touch WAN or 192.168.88.0/24.""" from __future__ import annotations import binascii import hashlib import socket import sys HOST = "192.168.88.1" PORT = 8728 USER = "admin" PASSWORD = "" def encode_len(n: int) -> bytes: if n < 0x80: return bytes([n]) if n < 0x4000: n |= 0x8000 return bytes([(n >> 8) & 0xFF, n & 0xFF]) raise RuntimeError("word too long") def write_sentence(sock: socket.socket, words: list[str]) -> None: for word in words: b = word.encode() sock.sendall(encode_len(len(b)) + b) sock.sendall(b"\x00") def read_len(sock: socket.socket) -> int: c = sock.recv(1) if not c: raise ConnectionError("closed") n = c[0] if n & 0x80 == 0: return n if n & 0xC0 == 0x80: n &= ~0xC0 b = sock.recv(1) return (n << 8) + b[0] raise RuntimeError("len") def read_sentence(sock: socket.socket) -> list[str]: words = [] while True: n = read_len(sock) if n == 0: break data = b"" while len(data) < n: chunk = sock.recv(n - len(data)) if not chunk: raise ConnectionError("closed") data += chunk words.append(data.decode("utf-8", "replace")) return words def talk(sock: socket.socket, words: list[str]) -> list[list[str]]: write_sentence(sock, words) out = [] while True: s = read_sentence(sock) if not s: break out.append(s) if s[0] in ("!done", "!trap", "!fatal"): break return out def ok(replies: list[list[str]]) -> bool: return any(s and s[0] == "!done" for s in replies) and not any(s and s[0] == "!trap" for s in replies) def trap_msg(replies: list[list[str]]) -> str: for s in replies: for w in s: if w.startswith("=message="): return w[9:] return str(replies) def has_name(sock: socket.socket, path: str, name: str, key: str = "name") -> bool: replies = talk(sock, [path, f"?{key}={name}"]) return any(s and s[0] == "!re" for s in replies) def add(sock: socket.socket, path: str, args: dict[str, str], exists_name: str | None = None, exists_key: str = "name") -> None: if exists_name and has_name(sock, path.replace("/add", "/print"), exists_name, exists_key): print("SKIP", path, exists_name) return words = [path] + [f"={k}={v}" for k, v in args.items()] r = talk(sock, words) if ok(r): print("OK ", path, exists_name or args) else: print("FAIL", path, exists_name or args, trap_msg(r)) raise SystemExit(1) def login(sock: socket.socket) -> None: r = talk(sock, ["/login", f"=name={USER}", f"=password={PASSWORD}"]) chal = None for s in r: for w in s: if w.startswith("=ret="): chal = w[5:] if chal: h = hashlib.md5() h.update(b"\x00" + PASSWORD.encode() + binascii.unhexlify(chal)) r = talk(sock, ["/login", f"=name={USER}", "=response=00" + h.hexdigest()]) if not ok(r): raise RuntimeError("login failed " + trap_msg(r)) def main() -> None: sock = socket.create_connection((HOST, PORT), 10) login(sock) print("LOGGED IN") r = talk(sock, ["/system/backup/save", "=name=woyla-before-hotspot"]) print("BACKUP", "OK" if ok(r) else trap_msg(r)) add(sock, "/interface/wireless/security-profiles/add", { "name": "woyla-open", "mode": "none", }, "woyla-open") add(sock, "/interface/wireless/add", { "name": "wlan-hotspot", "master-interface": "wlan1", "ssid": "Woyla-WiFi", "security-profile": "woyla-open", "disabled": "no", }, "wlan-hotspot") add(sock, "/interface/bridge/add", {"name": "bridge-hotspot"}, "bridge-hotspot") # port: check by interface name if not has_name(sock, "/interface/bridge/port/print", "wlan-hotspot", "interface"): add(sock, "/interface/bridge/port/add", { "bridge": "bridge-hotspot", "interface": "wlan-hotspot", }) else: print("SKIP /interface/bridge/port wlan-hotspot") if not has_name(sock, "/ip/address/print", "10.10.10.1/24", "address"): add(sock, "/ip/address/add", { "address": "10.10.10.1/24", "interface": "bridge-hotspot", "comment": "woyla-test-hotspot", }) else: print("SKIP address 10.10.10.1/24") add(sock, "/ip/pool/add", { "name": "hs-pool", "ranges": "10.10.10.10-10.10.10.200", }, "hs-pool") if not has_name(sock, "/ip/dhcp-server/network/print", "10.10.10.0/24", "address"): add(sock, "/ip/dhcp-server/network/add", { "address": "10.10.10.0/24", "gateway": "10.10.10.1", "dns-server": "10.10.10.1", "comment": "woyla-test-hotspot", }) else: print("SKIP dhcp network 10.10.10.0/24") add(sock, "/ip/dhcp-server/add", { "name": "dhcp-hotspot", "interface": "bridge-hotspot", "address-pool": "hs-pool", "disabled": "no", }, "dhcp-hotspot") add(sock, "/ip/hotspot/profile/add", { "name": "hs-woyla", "hotspot-address": "10.10.10.1", "dns-name": "hotspot.woyla", "html-directory": "hotspot", "login-by": "http-chap,http-pap,cookie", }, "hs-woyla") add(sock, "/ip/hotspot/add", { "name": "hotspot1", "interface": "bridge-hotspot", "address-pool": "hs-pool", "profile": "hs-woyla", "disabled": "no", }, "hotspot1") for host in [ "billing.woyla.net", "*.woyla.net", "mesomb.hachther.com", "*.mesomb.hachther.com", "business.mesomb.com", "*.mesomb.com", "*.cloudflare.com", "cloudflare.com", ]: if not has_name(sock, "/ip/hotspot/walled-garden/print", host, "dst-host"): add(sock, "/ip/hotspot/walled-garden/add", { "dst-host": host, "action": "allow", "comment": "woyla-test", }) else: print("SKIP walled-garden", host) for ip in ["192.168.88.21", "1.1.1.1", "8.8.8.8"]: if not has_name(sock, "/ip/hotspot/walled-garden/ip/print", ip, "dst-address"): add(sock, "/ip/hotspot/walled-garden/ip/add", { "dst-address": ip, "action": "accept", "comment": "woyla-test", }) else: print("SKIP walled-garden-ip", ip) for profile, rate in [ ("TEST-1H", "2M/2M"), ("TEST-1J", "4M/4M"), ("TEST-5GO", "5M/5M"), ]: add(sock, "/ip/hotspot/user/profile/add", { "name": profile, "shared-users": "1", "rate-limit": rate, "idle-timeout": "none", "keepalive-timeout": "2m", }, profile) print("\n==== VERIFY ====") for cmd in [ ["/interface/wireless/print", "?name=wlan-hotspot"], ["/ip/hotspot/print"], ["/ip/address/print", "?comment=woyla-test-hotspot"], ["/ip/route/print", "?dst-address=0.0.0.0/0"], ["/ip/address/print", "?address=192.168.88.1/24"], ]: print(cmd[0], cmd[1:] if len(cmd) > 1 else "") for s in talk(sock, cmd): print(" ", " ".join(s)[:220]) sock.close() print("DONE") if __name__ == "__main__": main()