refactor(router): Reboot the router over HTTPS with stdlib instead of Playwright
This commit is contained in:
1 parent
3b81216892
commit
6294dc0555
1 file changed
+76
-80
+76
-80
@@ -1,97 +1,93 @@
|
|||||||
#!/usr/bin/env -S uv run --script
|
#!/usr/bin/env python3
|
||||||
# /// script
|
|
||||||
# dependencies = [
|
|
||||||
# "playwright>=1.45.0",
|
|
||||||
# ]
|
|
||||||
# ///
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
import sys
|
import base64
|
||||||
import time
|
import json
|
||||||
|
import logging
|
||||||
|
import ssl
|
||||||
|
import urllib.parse
|
||||||
|
import urllib.request
|
||||||
|
|
||||||
from playwright.sync_api import Playwright, TimeoutError as PlaywrightTimeoutError, sync_playwright
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
ROUTER_URL = "http://192.168.1.1"
|
ROUTER_HOST = "192.168.1.1"
|
||||||
ROUTER_USERNAME = "abdus"
|
ROUTER_USERNAME = "abdus"
|
||||||
ROUTER_PASSWORD = "xAsametk50"
|
ROUTER_PASSWORD = "xAsametk50"
|
||||||
|
|
||||||
|
|
||||||
def reboot_router(playwright: Playwright, headed: bool, timeout_ms: int) -> None:
|
|
||||||
browser = playwright.chromium.launch(headless=not headed)
|
|
||||||
context = browser.new_context(ignore_https_errors=True)
|
|
||||||
page = context.new_page()
|
|
||||||
page.set_default_timeout(timeout_ms)
|
|
||||||
|
|
||||||
try:
|
|
||||||
page.goto(f"{ROUTER_URL}/login", wait_until="domcontentloaded")
|
|
||||||
|
|
||||||
username_box = page.get_by_role("textbox", name="User Name")
|
|
||||||
password_box = page.get_by_role("textbox", name="Password")
|
|
||||||
|
|
||||||
username_box.click()
|
|
||||||
username_box.press("ControlOrMeta+a")
|
|
||||||
username_box.press("Backspace")
|
|
||||||
username_box.type(ROUTER_USERNAME, delay=80)
|
|
||||||
|
|
||||||
password_box.click()
|
|
||||||
password_box.press("ControlOrMeta+a")
|
|
||||||
password_box.press("Backspace")
|
|
||||||
password_box.type(ROUTER_PASSWORD, delay=140)
|
|
||||||
page.get_by_role("textbox", name="Password").press("Enter")
|
|
||||||
|
|
||||||
page.wait_for_load_state("domcontentloaded")
|
|
||||||
|
|
||||||
try:
|
|
||||||
page.get_by_text("The username or password is not correct", exact=False).wait_for(
|
|
||||||
state="visible", timeout=1500
|
|
||||||
)
|
|
||||||
raise RuntimeError("Router rejected configured credentials")
|
|
||||||
except PlaywrightTimeoutError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
menu_trigger = page.locator("#h_menu_list").first
|
|
||||||
if menu_trigger.count() and menu_trigger.is_visible():
|
|
||||||
menu_trigger.click()
|
|
||||||
|
|
||||||
restart_icon = page.locator("#navbar_reboot .icon-menu-restart, #navbar_reboot").first
|
|
||||||
restart_icon.wait_for(state="visible")
|
|
||||||
|
|
||||||
with page.expect_response(lambda r: "/cgi-bin/Reboot" in r.url, timeout=timeout_ms):
|
|
||||||
restart_icon.click()
|
|
||||||
page.get_by_role("button", name="OK").click()
|
|
||||||
|
|
||||||
time.sleep(5)
|
|
||||||
|
|
||||||
print("Router reboot request sent.")
|
|
||||||
finally:
|
|
||||||
context.close()
|
|
||||||
browser.close()
|
|
||||||
|
|
||||||
|
|
||||||
def parse_args() -> argparse.Namespace:
|
def parse_args() -> argparse.Namespace:
|
||||||
parser = argparse.ArgumentParser(description="Log in and reboot router")
|
parser = argparse.ArgumentParser(
|
||||||
parser.add_argument("--headed", action="store_true", help="Run with visible browser")
|
description="Log in to the Zyxel router over HTTPS and reboot it",
|
||||||
parser.add_argument(
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||||||
"--timeout",
|
|
||||||
type=int,
|
|
||||||
default=60000,
|
|
||||||
help="Timeout in milliseconds (default: 60000)",
|
|
||||||
)
|
)
|
||||||
|
parser.add_argument("--host", default=ROUTER_HOST)
|
||||||
|
parser.add_argument("--username", default=ROUTER_USERNAME)
|
||||||
|
parser.add_argument("--timeout", type=float, default=15.0, help="Request timeout in seconds")
|
||||||
|
parser.add_argument("--dry-run", action="store_true", help="Log in but do not reboot")
|
||||||
return parser.parse_args()
|
return parser.parse_args()
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
def build_opener() -> urllib.request.OpenerDirector:
|
||||||
|
# The router uses a self-signed certificate. Over HTTPS the UI skips its RSA/AES login layer.
|
||||||
|
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 post(opener: urllib.request.OpenerDirector, url: str, body: dict, timeout: float) -> dict:
|
||||||
|
request = urllib.request.Request(
|
||||||
|
url=url,
|
||||||
|
data=json.dumps(body).encode(),
|
||||||
|
headers={"Content-Type": "application/json"},
|
||||||
|
method="POST",
|
||||||
|
)
|
||||||
|
with opener.open(request, timeout=timeout) as response:
|
||||||
|
raw = response.read().decode()
|
||||||
|
return json.loads(raw) if raw.strip() else {}
|
||||||
|
|
||||||
|
|
||||||
|
def login(opener: urllib.request.OpenerDirector, base_url: str, username: str, timeout: float) -> str:
|
||||||
|
body = post(
|
||||||
|
opener=opener,
|
||||||
|
url=f"{base_url}/UserLogin",
|
||||||
|
body={
|
||||||
|
"Input_Account": username,
|
||||||
|
"Input_Passwd": base64.b64encode(ROUTER_PASSWORD.encode()).decode(),
|
||||||
|
"currLang": "en",
|
||||||
|
"RememberPassword": 0,
|
||||||
|
"SHA512_password": False,
|
||||||
|
},
|
||||||
|
timeout=timeout,
|
||||||
|
)
|
||||||
|
if body.get("result") != "ZCFG_SUCCESS":
|
||||||
|
raise RuntimeError(f"Login rejected: {body}")
|
||||||
|
return body.get("sessionkey", "")
|
||||||
|
|
||||||
|
|
||||||
|
def reboot(opener: urllib.request.OpenerDirector, base_url: str, session_key: str, timeout: float) -> None:
|
||||||
|
query = f"?{urllib.parse.urlencode({'sessionkey': session_key})}" if session_key else ""
|
||||||
|
body = post(opener=opener, url=f"{base_url}/cgi-bin/Reboot{query}", body={}, timeout=timeout)
|
||||||
|
logger.info("Reboot response: %s", body)
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
args = parse_args()
|
args = parse_args()
|
||||||
try:
|
base_url = f"https://{args.host}"
|
||||||
with sync_playwright() as playwright:
|
opener = build_opener()
|
||||||
reboot_router(playwright, headed=args.headed, timeout_ms=args.timeout)
|
|
||||||
return 0
|
session_key = login(opener=opener, base_url=base_url, username=args.username, timeout=args.timeout)
|
||||||
except Exception as exc:
|
logger.info("Logged in as %s", args.username)
|
||||||
print(f"Error: {exc}", file=sys.stderr)
|
if args.dry_run:
|
||||||
return 1
|
logger.info("Dry run, not rebooting")
|
||||||
|
return
|
||||||
|
reboot(opener=opener, base_url=base_url, session_key=session_key, timeout=args.timeout)
|
||||||
|
logger.info("Reboot request sent, router is back in about a minute")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
raise SystemExit(main())
|
main()
|
||||||
Reference in new issue
Block a user