python/check_router.py - master - monoceros/hilink-api

python/check_router.py
440 строк16 KB
Алексей Сопов
first_commit
12 янв 2026, 18:17
12 янв 2026, 18:17845f41c
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Проверка состояния роутера Huawei (HiLink): доступность, SIM, сеть и готовность SMS API.

Настройки берутся из переменных окружения:
- HUAWEI_URL
- HUAWEI_USERNAME
- HUAWEI_PASSWORD

CLI-аргументы могут переопределять значения из окружения.
"""

import argparse
import json
import os
import time
from urllib.parse import urljoin
import xml.etree.ElementTree as ET

import requests
import urllib3

from env_loader import load_env
from hilink_scram import generate_nonce, scram_client_proof


def safe_print(text: str) -> None:
    print(text, flush=True)


def _xml_to_obj(elem: ET.Element):
    children = list(elem)
    if not children:
        return (elem.text or "").strip()

    out: dict[str, object] = {}
    for c in children:
        key = c.tag
        val = _xml_to_obj(c)
        if key in out:
            existing = out[key]
            if isinstance(existing, list):
                existing.append(val)
            else:
                out[key] = [existing, val]
        else:
            out[key] = val
    return out


def parse_xml_response(content: bytes) -> dict:
    try:
        root = ET.fromstring(content)
    except ET.ParseError:
        return {"ok": False, "error": "invalid_xml"}

    if root.tag == "error":
        return {"ok": False, "error": "api_error", "code": root.findtext("code")}

    return {"ok": True, "data": _xml_to_obj(root)}


class HuaweiClient:
    def __init__(self, base_url: str, username: str, password: str, debug: bool = False):
        self.base_url = base_url.rstrip("/")
        self.username = username
        self.password = password
        self.debug = debug

        self.session = requests.Session()
        self.session.verify = False
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

        self.session_id: str | None = None
        self.token: str | None = None

    def _dlog(self, msg: str) -> None:
        if self.debug:
            safe_print(msg)

    def process_http_headers(self, response: requests.Response) -> None:
        if "SessionID" in response.cookies:
            self.session_id = response.cookies["SessionID"]
            self.session.cookies.set("SessionID", self.session_id)

        if "__RequestVerificationToken" in response.headers:
            new_token = response.headers["__RequestVerificationToken"].split("#")[0]
            if new_token:
                self.token = new_token

    def _req(self, method: str, path: str, data: str | None = None, *, add_ajax_headers: bool = False) -> dict:
        url = urljoin(self.base_url, path)
        headers: dict[str, str] = {
            "Accept": "*/*",
        }

        if self.session_id:
            headers["Cookie"] = f"SessionID={self.session_id}"
        if self.token:
            headers["__RequestVerificationToken"] = self.token
        if add_ajax_headers:
            headers.update(
                {
                    "X-Requested-With": "XMLHttpRequest",
                    "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
                }
            )

        self._dlog(f"[DEBUG] {method} {path}")

        try:
            r = self.session.request(method, url, data=data, headers=headers, timeout=10)
            self.process_http_headers(r)
        except Exception as e:
            return {"ok": False, "error": "request_failed", "detail": str(e)}

        out = {"http_status": r.status_code}
        if r.status_code != 200:
            out.update({"ok": False, "error": "http_error"})
            return out

        parsed = parse_xml_response(r.content)
        out.update(parsed)
        return out

    def get_sestokinfo(self) -> dict:
        return self._req("GET", "/api/webserver/SesTokInfo")

    def scram_login(self, retries: int = 3) -> dict:
        last: dict | None = None
        for attempt in range(1, retries + 1):
            last = self._scram_login_once()
            if last.get("ok"):
                return last
            self._dlog(f"[DEBUG] SCRAM login failed (attempt {attempt}/{retries})")
            time.sleep(0.25 * attempt)
        return last or {"ok": False, "error": "scram_failed"}

    def _scram_login_once(self) -> dict:
        # 1) setup session
        r = self._req("GET", "/")
        if not r.get("http_status"):
            return {"ok": False, "error": "request_failed"}

        # 2) server token
        tok = self._req("GET", "/api/webserver/token")
        if not tok.get("ok"):
            return {"ok": False, "error": "token_failed", "detail": tok}
        token_val = None
        try:
            token_val = tok["data"]["token"]  # type: ignore[index]
        except Exception:
            pass
        if not isinstance(token_val, str) or not token_val:
            return {"ok": False, "error": "token_missing"}

        verification_token = token_val[32:] if len(token_val) > 32 else token_val

        # 3) challenge_login
        clientnonce = generate_nonce()
        challenge_xml = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            "<request>"
            f"<username>{self.username}</username>"
            f"<firstnonce>{clientnonce}</firstnonce>"
            "<mode>1</mode>"
            "</request>"
        )

        url = urljoin(self.base_url, "/api/user/challenge_login")
        headers = {"Content-type": "text/html", "__RequestVerificationToken": verification_token}
        try:
            rr = self.session.post(url, data=challenge_xml, headers=headers, timeout=10)
            self.process_http_headers(rr)
        except Exception as e:
            return {"ok": False, "error": "challenge_failed", "detail": str(e)}

        if rr.status_code != 200:
            return {"ok": False, "error": "challenge_http", "http_status": rr.status_code}

        ch = parse_xml_response(rr.content)
        if not ch.get("ok"):
            return {"ok": False, "error": "challenge_api", "detail": ch}

        data = ch.get("data", {})
        if not isinstance(data, dict):
            return {"ok": False, "error": "challenge_parse"}

        servernonce = str(data.get("servernonce") or "")
        salt_hex = str(data.get("salt") or "")
        try:
            iterations = int(str(data.get("iterations") or "0"))
        except Exception:
            iterations = 0
        if not servernonce or not salt_hex or iterations <= 0:
            return {"ok": False, "error": "challenge_params"}

        verification_token = rr.headers.get("__RequestVerificationToken", "").split("#")[0] or verification_token

        # 4) authentication_login
        clientproof = scram_client_proof(self.password, clientnonce, servernonce, salt_hex, iterations)
        auth_xml = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            "<request>"
            f"<clientproof>{clientproof}</clientproof>"
            f"<finalnonce>{servernonce}</finalnonce>"
            "</request>"
        )

        url = urljoin(self.base_url, "/api/user/authentication_login")
        headers = {
            "Content-type": "application/x-www-form-urlencoded; charset=UTF-8",
            "__RequestVerificationToken": verification_token,
            "X-Requested-With": "XMLHttpRequest",
        }
        try:
            rr = self.session.post(url, data=auth_xml, headers=headers, timeout=10)
            self.process_http_headers(rr)
        except Exception as e:
            return {"ok": False, "error": "auth_failed", "detail": str(e)}

        if rr.status_code != 200:
            return {"ok": False, "error": "auth_http", "http_status": rr.status_code}

        auth = parse_xml_response(rr.content)
        if not auth.get("ok"):
            return {"ok": False, "error": "auth_api", "detail": auth}

        tok_one = rr.headers.get("__RequestVerificationTokenone")
        tok_std = rr.headers.get("__RequestVerificationToken")
        tok = (tok_one or tok_std or "").split("#")[0]
        if tok:
            self.token = tok

        # обновляем SesTokInfo, чтобы дальше использовать TokInfo
        self.get_sestokinfo()
        return {"ok": True}

    def monitoring_status(self) -> dict:
        return self._req("GET", "/api/monitoring/status")

    def monitoring_signal(self) -> dict:
        return self._req("GET", "/api/monitoring/signal")

    def check_notifications(self) -> dict:
        return self._req("GET", "/api/monitoring/check-notifications")

    def pin_status(self) -> dict:
        return self._req("GET", "/api/pin/status")

    def current_plmn(self) -> dict:
        return self._req("GET", "/api/net/current-plmn")

    def sms_count(self) -> dict:
        return self._req("GET", "/api/sms/sms-count")

    def sms_send_status(self) -> dict:
        return self._req("GET", "/api/sms/send-status")


def mask_phone(value: str) -> str:
    v = (value or "").strip()
    if not v:
        return v
    # оставляем + и последние 2-4 цифры
    digits = "".join(ch for ch in v if ch.isdigit())
    if len(digits) <= 4:
        return v
    tail = digits[-4:]
    return ("+" if v.startswith("+") else "") + ("*" * max(0, len(digits) - 4)) + tail


SENSITIVE_KEYS = {"TokInfo", "SesInfo", "SessionID", "__RequestVerificationToken"}
SENSITIVE_PHONE_KEYS = {"Phone", "SucPhone", "FailPhone", "phone"}


def redact_obj(obj):
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            if k in SENSITIVE_KEYS:
                out[k] = "<redacted>"
                continue
            if k in SENSITIVE_PHONE_KEYS and isinstance(v, str):
                out[k] = mask_phone(v)
                continue
            out[k] = redact_obj(v)
        return out
    if isinstance(obj, list):
        return [redact_obj(x) for x in obj]
    return obj


def _sim_state_human(data: dict) -> str | None:
    # Часто встречаются поля: SimState, PinOptState, SimLockStatus
    sim_state = data.get("SimState")
    if sim_state is None:
        return None
    sim_state_s = str(sim_state)
    mapping = {
        "0": "SIM отсутствует/не определена",
        "1": "SIM готова",
        "2": "Требуется PIN",
        "3": "Требуется PUK",
        "4": "SIM заблокирована",
    }
    return mapping.get(sim_state_s, f"SimState={sim_state_s}")


def main() -> int:
    # Подхватываем переменные из .env (если файл есть и установлен python-dotenv)
    load_env()

    p = argparse.ArgumentParser(description="Проверка состояния Huawei HiLink роутера (SIM/сеть/SMS)")
    p.add_argument("--url", default=os.getenv("HUAWEI_URL"), help="URL роутера (или env HUAWEI_URL)")
    p.add_argument("--username", default=os.getenv("HUAWEI_USERNAME"), help="Логин (или env HUAWEI_USERNAME)")
    p.add_argument("--password", default=os.getenv("HUAWEI_PASSWORD"), help="Пароль (или env HUAWEI_PASSWORD)")
    p.add_argument("--json", action="store_true", help="Вывод JSON")
    p.add_argument("--raw", action="store_true", help="JSON без редактирования чувствительных полей")
    p.add_argument("--debug", action="store_true", help="Подробные логи")
    args = p.parse_args()

    if not args.url or not args.username or not args.password:
        safe_print("[ERR] Missing router credentials.")
        safe_print("Set env vars: HUAWEI_URL, HUAWEI_USERNAME, HUAWEI_PASSWORD (or pass --url/--username/--password).")
        return 2

    client = HuaweiClient(args.url, args.username, args.password, debug=args.debug)

    report: dict[str, object] = {
        "ok": True,
        "url": args.url,
        "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
        "checks": {},
        "sms_ready": {},
        "errors": [],
    }

    checks: dict[str, object] = {}
    report["checks"] = checks

    # базовая доступность + initial SesTokInfo
    sestok = client.get_sestokinfo()
    checks["sestokinfo"] = sestok
    if not sestok.get("ok"):
        report["ok"] = False

    # статусы без логина (если доступно на прошивке)
    for name, fn in (
        ("monitoring_status", client.monitoring_status),
        ("pin_status", client.pin_status),
        ("monitoring_signal", client.monitoring_signal),
        ("current_plmn", client.current_plmn),
        ("sms_count", client.sms_count),
        ("check_notifications", client.check_notifications),
    ):
        res = fn()
        checks[name] = res

    # готовность SMS: SCRAM + доступ к send-status
    scram = client.scram_login(retries=3)
    sms_ready: dict[str, object] = {"scram": scram}
    report["sms_ready"] = sms_ready
    if not scram.get("ok"):
        report["ok"] = False
    else:
        # после SCRAM часть эндпоинтов может стать доступнее
        checks["monitoring_status_auth"] = client.monitoring_status()
        checks["monitoring_signal_auth"] = client.monitoring_signal()
        sms_ready["send_status"] = client.sms_send_status()
        sms_ready["sms_count"] = client.sms_count()
        if not sms_ready["send_status"].get("ok"):
            report["ok"] = False

    # человекочитаемый вывод
    if args.json:
        out = report if args.raw else redact_obj(report)
        safe_print(json.dumps(out, ensure_ascii=False, indent=2))
        return 0 if report.get("ok") else 1

    safe_print(f"[OK] url={args.url}" if report.get("ok") else f"[ERR] url={args.url}")

    pin = checks.get("pin_status", {})
    pin_data = None
    if isinstance(pin, dict) and pin.get("ok") and isinstance(pin.get("data"), dict):
        pin_data = pin["data"]
    if isinstance(pin_data, dict):
        sim_h = _sim_state_human(pin_data)
        if sim_h:
            safe_print(f"SIM: {sim_h}")

    mon = checks.get("monitoring_status", {})
    mon_data = None
    if isinstance(mon, dict) and mon.get("ok") and isinstance(mon.get("data"), dict):
        mon_data = mon["data"]
    if isinstance(mon_data, dict):
        cs = mon_data.get("ConnectionStatus")
        nt = mon_data.get("CurrentNetworkType")
        ip = mon_data.get("WanIPAddress")
        parts = []
        if cs is not None:
            parts.append(f"ConnectionStatus={cs}")
        if nt is not None:
            parts.append(f"NetworkType={nt}")
        if ip:
            parts.append(f"WAN={ip}")
        if parts:
            safe_print("Сеть: " + ", ".join(parts))

    sig = checks.get("monitoring_signal", {})
    sig_data = None
    if isinstance(sig, dict) and sig.get("ok") and isinstance(sig.get("data"), dict):
        sig_data = sig["data"]
    if isinstance(sig_data, dict):
        keys = ("rsrp", "rsrq", "sinr", "rssi", "cell_id")
        parts = []
        for k in keys:
            v = sig_data.get(k.upper()) or sig_data.get(k)  # иногда поля в верхнем регистре
            if v:
                parts.append(f"{k}={v}")
        if parts:
            safe_print("Сигнал: " + ", ".join(parts))

    sms = report.get("sms_ready", {})
    if isinstance(sms, dict):
        scram_ok = bool(isinstance(sms.get("scram"), dict) and sms["scram"].get("ok"))
        send_ok = bool(isinstance(sms.get("send_status"), dict) and sms["send_status"].get("ok"))
        if scram_ok and send_ok:
            safe_print("SMS API: готово")
        elif scram_ok:
            safe_print("SMS API: SCRAM ok, но send-status недоступен")
        else:
            safe_print("SMS API: не готово (SCRAM/login)")

    return 0 if report.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())