127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
import os
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
BASE_URL = os.environ.get("SWITCH_URL", "http://192.168.2.10")
|
|
SESSION_COOKIE_FILE = os.path.join(os.path.dirname(__file__), ".session_cookie")
|
|
|
|
|
|
def login(password: str = "", session: requests.Session | None = None) -> requests.Session:
|
|
"""POST to /hp_login.html, return a Session with the SID cookie set."""
|
|
session = session or requests.Session()
|
|
# A bare POST occasionally comes back without setting SID (the switch
|
|
# seems to want a session cookie already present, even an empty one,
|
|
# before it will issue a real one) -- a GET first makes login reliable.
|
|
session.get(f"{BASE_URL}/")
|
|
resp = session.post(
|
|
f"{BASE_URL}/hp_login.html",
|
|
data={"pwd": password, "login": "Login", "err_flag": "", "err_msg": ""},
|
|
)
|
|
resp.raise_for_status()
|
|
if "<title>login</title>" in resp.text.lower():
|
|
raise RuntimeError(
|
|
"Login failed, response still looks like the login page. "
|
|
f"Body snippet: {resp.text[:500]!r}"
|
|
)
|
|
# The switch only serves the main frameset as the direct response to this
|
|
# POST — a later GET "/" re-renders the login page regardless of session
|
|
# state, so stash it here for menu.get_treeframe_url() to use.
|
|
session.frameset_html = resp.text
|
|
return session
|
|
|
|
|
|
def _load_cached_session() -> requests.Session | None:
|
|
try:
|
|
with open(SESSION_COOKIE_FILE) as f:
|
|
sid = f.read().strip()
|
|
except FileNotFoundError:
|
|
return None
|
|
if not sid:
|
|
return None
|
|
session = requests.Session()
|
|
session.cookies.set("SID", sid, domain=urlparse(BASE_URL).hostname, path="/")
|
|
return session
|
|
|
|
|
|
def _is_logged_in(session: requests.Session) -> bool:
|
|
resp = session.get(f"{BASE_URL}/SysDescription.html")
|
|
resp.raise_for_status()
|
|
return "<title>login</title>" not in resp.text.lower()
|
|
|
|
|
|
def _cache_session(session: requests.Session) -> None:
|
|
sid = session.cookies.get("SID")
|
|
if sid:
|
|
with open(SESSION_COOKIE_FILE, "w") as f:
|
|
f.write(sid)
|
|
|
|
|
|
def get_session(password: str = "", force_login: bool = False) -> requests.Session:
|
|
"""Reuse a still-valid cached session if one exists, otherwise log in.
|
|
|
|
The switch only allows one active web session at a time and briefly
|
|
refuses a fresh login right after a previous one (its ~5 minute session
|
|
timeout hasn't elapsed yet), so avoiding a needless re-login here is
|
|
what makes running a script repeatedly in short succession reliable.
|
|
"""
|
|
if not force_login:
|
|
cached = _load_cached_session()
|
|
if cached is not None and _is_logged_in(cached):
|
|
return cached
|
|
|
|
session = login(password)
|
|
_cache_session(session)
|
|
return session
|
|
|
|
|
|
def _resolve_url(path: str) -> str:
|
|
if path.startswith("http://") or path.startswith("https://"):
|
|
return path
|
|
if path.startswith("/"):
|
|
return f"{BASE_URL}{path}"
|
|
return f"{BASE_URL}/{path}"
|
|
|
|
|
|
def fetch(session: requests.Session, path: str) -> str:
|
|
"""GET BASE_URL + path (path may be relative or absolute), return response.text."""
|
|
resp = session.get(_resolve_url(path))
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
|
|
|
|
def submit_form(session: requests.Session, path: str, fields: dict) -> str:
|
|
"""POST fields to path, merged with the firmware's standard bookkeeping
|
|
fields (submit_flag, submit_target, err_flag, err_msg, clazz_information).
|
|
|
|
Raises RuntimeError if the response's err_flag comes back "1" -- the
|
|
firmware's own validation-failure signal. NOTE: most value validation in
|
|
this firmware happens client-side in JS, not on the server -- an invalid
|
|
enum value, for example, is just silently ignored (state left unchanged)
|
|
rather than reported via err_flag. So this check catches *some* rejected
|
|
writes, but a caller that needs certainty should read the value back
|
|
afterwards and compare, rather than trusting the absence of an error.
|
|
"""
|
|
target = path[1:] if path.startswith("/") else path
|
|
data = {
|
|
# 8 == xui_operation_submit (per the firmware's own _xe_jsvars.js);
|
|
# 1 is xui_operation_reload and silently applies nothing.
|
|
"submit_flag": "8",
|
|
"submit_target": target,
|
|
"err_flag": "0",
|
|
"err_msg": "",
|
|
"clazz_information": target,
|
|
}
|
|
data.update(fields)
|
|
resp = session.post(_resolve_url(path), data=data)
|
|
resp.raise_for_status()
|
|
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
err_flag = soup.find("input", attrs={"name": "err_flag"})
|
|
if err_flag is not None and err_flag.get("value") == "1":
|
|
err_msg = soup.find("input", attrs={"name": "err_msg"})
|
|
msg = err_msg.get("value") if err_msg is not None else "unknown error"
|
|
raise RuntimeError(f"Switch rejected write to {path!r}: {msg}")
|
|
return resp.text
|