85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# dependencies = [
|
|
# "httpx>=0.27.0",
|
|
# ]
|
|
# ///
|
|
|
|
import argparse
|
|
import time
|
|
|
|
import httpx
|
|
|
|
ROUTER_BASE_URL = "http://192.168.1.1"
|
|
|
|
# Captured from successful login flow for abdus:xAsametk50.
|
|
# Firmware expects encrypted payload at /UserLogin.
|
|
USER_LOGIN_PAYLOAD = {
|
|
"content": "z7oqX7AWMrPjJdAEgrXCZY9q/+mrsyodlD53apbPmPXBkuB1SJXdoXr8+NcUbuRwn/jyc3HJlyKcMrAHDl3aFbN8/Gnyy8TzRKQkCAoSdVEVifzmWBPgpoDKvvT/p6n4TrAi2+b1eJ18Sr4ab1MWrI0XKNbBLEU+0UJlAd/UOHo=",
|
|
"key": "UwLW2sfQZ5S5XegolmSX0kqpxwzeJEIXq8lp3gLkNpNX+qiBAC4E2ColaigMtsMPbVEEFGk+RiPixF9L4xI4licoguc71zEfG9U3YUvNZZeBpV3JKMLLDzhs5aFDIb9M4k/zQ623zByOWDKjnUhHqqRK1I8hEyhThbbhKUkvo67/6vcRHHE9c0Lcxc2SrIGeyRUvPe5F+zXH8PbUn4aPq6fVTPmECRbiZOOm6zoy99PvyRVIYiCOkWqeXHhGIdMl2YZhA2UeM+1lwWtFQjSgefPeCfdhtIuEfFfUUaj3DyPJfD/+CyUT0QGUH6SD5sf1M8DySqOglgH1dIXEPLyIwA==",
|
|
"iv": "u+z76g62wnZYAehSfgffrPzpstykp6E891Z9CrDLfO0=",
|
|
}
|
|
|
|
|
|
def _extract_session_key(resp: httpx.Response) -> int:
|
|
data = resp.json()
|
|
if isinstance(data, list) and data:
|
|
return int(data[0]["SessionKey"])
|
|
if isinstance(data, dict) and "SessionKey" in data:
|
|
return int(data["SessionKey"])
|
|
raise RuntimeError(f"Could not parse SessionKey from /changeSessionKey response: {data!r}")
|
|
|
|
|
|
def _wait_for_router(client: httpx.Client, max_wait: float) -> None:
|
|
deadline = time.time() + max_wait
|
|
last_error = None
|
|
while time.time() < deadline:
|
|
try:
|
|
resp = client.get("/login")
|
|
resp.raise_for_status()
|
|
return
|
|
except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.ConnectError, httpx.NetworkError) as exc:
|
|
last_error = exc
|
|
time.sleep(2)
|
|
raise RuntimeError(f"Router not reachable within {max_wait:.0f}s") from last_error
|
|
|
|
|
|
def reboot_router(timeout: float, post_confirm_sleep: float) -> None:
|
|
with httpx.Client(base_url=ROUTER_BASE_URL, timeout=timeout, follow_redirects=True) as client:
|
|
_wait_for_router(client, max_wait=90)
|
|
|
|
login = client.post("/UserLogin", json=USER_LOGIN_PAYLOAD)
|
|
login.raise_for_status()
|
|
|
|
check = client.get("/cgi-bin/UserLoginCheck")
|
|
check.raise_for_status()
|
|
if "ZCFG_SUCCESS" not in check.text:
|
|
raise RuntimeError(f"Login check failed: {check.text}")
|
|
|
|
key_resp = client.get("/changeSessionKey")
|
|
key_resp.raise_for_status()
|
|
session_key = _extract_session_key(key_resp)
|
|
|
|
reboot = client.post("/cgi-bin/Reboot", params={"sessionkey": session_key})
|
|
reboot.raise_for_status()
|
|
|
|
time.sleep(post_confirm_sleep)
|
|
print("Router reboot request sent.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Reboot router via HTTP API (no Playwright)")
|
|
parser.add_argument("--timeout", type=float, default=20.0, help="HTTP timeout in seconds")
|
|
parser.add_argument(
|
|
"--post-confirm-sleep",
|
|
type=float,
|
|
default=5.0,
|
|
help="Seconds to wait after reboot request",
|
|
)
|
|
args = parser.parse_args()
|
|
reboot_router(timeout=args.timeout, post_confirm_sleep=args.post_confirm_sleep)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|