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

python/read_sms.py
449 строк17 KB
Алексей Сопов
first_commit
12 янв 2026, 18:17
12 янв 2026, 18:17845f41c
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Чтение SMS через веб-интерфейс Huawei B618s-22d (HiLink).

Работает через SCRAM (challenge_login/authentication_login), затем читает сообщения через /api/sms/sms-list.
"""

import argparse
import json
import time
import os
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


BOX_TYPES = {
    "inbox": 1,
    "sent": 2,
    "draft": 3,
}


def safe_print(text: str) -> None:
    """Обычная печать."""
    print(text, flush=True)


class HuaweiHiLinkClient:
    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 = None
        self.token = None
    
    def _dlog(self, msg: str) -> None:
        if self.debug:
            safe_print(msg)

    def _dresp(self, label: str, r: requests.Response) -> None:
        if not self.debug:
            return
        try:
            body = r.content[:300]
        except Exception:
            body = b""
        safe_print(f"[DEBUG] {label}: status={getattr(r, 'status_code', None)}")
        if body:
            try:
                safe_print(f"[DEBUG] {label}: body={body.decode('utf-8', errors='replace')}")
            except Exception:
                safe_print(f"[DEBUG] {label}: body=<bytes {len(body)}>")

    def process_http_headers(self, response: requests.Response) -> None:
        """Обновляет SessionID и __RequestVerificationToken из ответа (как в HiLink API)."""
        if "SessionID" in response.cookies:
            self.session_id = response.cookies["SessionID"]
            self.session.cookies.set("SessionID", self.session_id)
            self._dlog(f"[DEBUG] SessionID updated: {self.session_id[:50]}...")

        if "__RequestVerificationToken" in response.headers:
            new_token = response.headers["__RequestVerificationToken"].split("#")[0]
            if new_token and new_token != self.token:
                self.token = new_token
                self._dlog(f"[DEBUG] Token updated: {self.token[:50]}...")

    def get_session_token(self) -> bool:
        """GET /api/webserver/SesTokInfo"""
        url = urljoin(self.base_url, "/api/webserver/SesTokInfo")
        r = self.session.get(url, timeout=10)
        self.process_http_headers(r)
        if r.status_code != 200:
            return False
        try:
            # Парсим из bytes, чтобы корректно учесть encoding="UTF-8" в XML
            root = ET.fromstring(r.content)
            ses = root.find("SesInfo")
            tok = root.find("TokInfo")
            if ses is not None and ses.text:
                self.session_id = ses.text
            if tok is not None and tok.text:
                self.token = tok.text
            return bool(self.session_id and self.token)
        except ET.ParseError:
            return False

    def scram_login(self) -> bool:
        """SCRAM login: /api/webserver/token -> /api/user/challenge_login -> /api/user/authentication_login"""
        try:
            # 1) setup session
            r = self.session.get(urljoin(self.base_url, "/"), timeout=10)
            self.process_http_headers(r)
            self._dresp("GET /", r)
            time.sleep(0.2)

            # 2) get server token
            r = self.session.get(urljoin(self.base_url, "/api/webserver/token"), timeout=10)
            self.process_http_headers(r)
            self._dresp("GET /api/webserver/token", r)
            if r.status_code != 200:
                return False
            time.sleep(0.15)

            root = ET.fromstring(r.content)
            token_elem = root.find("./token")
            if token_elem is None or not token_elem.text:
                return False

            server_token = token_elem.text
            verification_token = server_token[32:] if len(server_token) > 32 else server_token

            # 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>"
            )
            headers = {"Content-type": "text/html", "__RequestVerificationToken": verification_token}
            r = self.session.post(
                urljoin(self.base_url, "/api/user/challenge_login"),
                data=challenge_xml,
                headers=headers,
                timeout=10,
            )
            self.process_http_headers(r)
            self._dresp("POST /api/user/challenge_login", r)
            if r.status_code != 200:
                return False
            time.sleep(0.15)

            scram = ET.fromstring(r.content)
            servernonce_elem = scram.find("./servernonce")
            salt_elem = scram.find("./salt")
            iterations_elem = scram.find("./iterations")
            if servernonce_elem is None or salt_elem is None or iterations_elem is None:
                return False

            servernonce = servernonce_elem.text or ""
            salt_hex = salt_elem.text or ""
            iterations = int(iterations_elem.text or "0")
            if not servernonce or not salt_hex or iterations <= 0:
                return False

            verification_token = r.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>"
            )
            headers = {
                "Content-type": "application/x-www-form-urlencoded; charset=UTF-8",
                "__RequestVerificationToken": verification_token,
                "X-Requested-With": "XMLHttpRequest",
            }
            r = self.session.post(
                urljoin(self.base_url, "/api/user/authentication_login"),
                data=auth_xml,
                headers=headers,
                timeout=10,
            )
            self.process_http_headers(r)
            self._dresp("POST /api/user/authentication_login", r)
            if r.status_code != 200:
                return False

            tok_one = r.headers.get("__RequestVerificationTokenone")
            tok_std = r.headers.get("__RequestVerificationToken")
            tok = (tok_one or tok_std or "").split("#")[0]
            if tok:
                self.token = tok
                self._dlog(f"[DEBUG] SCRAM token: {tok[:50]}...")

            # Refresh SesTokInfo after SCRAM
            self.get_session_token()
            return bool(self.session_id and self.token)

        except Exception as e:
            self._dlog(f"[DEBUG] SCRAM login failed: {e}")
            return False

    def ensure_logged_in(self) -> bool:
        """Логин (SCRAM) для доступа к SMS API."""
        # Небольшой retry, чтобы не ловить редкие временные сбои
        delay = 0.3
        for attempt in range(1, 6):
            if self.scram_login():
                return True
            self._dlog(f"[DEBUG] SCRAM retry {attempt}/5 failed, sleeping {delay:.1f}s")
            time.sleep(delay)
            delay = min(delay * 2, 3.0)
        return False

    def sms_list(self, box_type: int, page_index: int, read_count: int, unread_preferred: int = 0) -> dict:
        """POST /api/sms/sms-list"""
        url = urljoin(self.base_url, "/api/sms/sms-list")
        referer = urljoin(self.base_url, "/html/smsinbox.html?smsinbox")

        xml_req = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            "<request>"
            f"<PageIndex>{page_index}</PageIndex>"
            f"<ReadCount>{read_count}</ReadCount>"
            f"<BoxType>{box_type}</BoxType>"
            "<SortType>0</SortType>"
            "<Ascending>0</Ascending>"
            f"<UnreadPreferred>{unread_preferred}</UnreadPreferred>"
            "</request>"
        )

        headers = {
            "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
            "X-Requested-With": "XMLHttpRequest",
            "X-KL-saas-Ajax-Request": "Ajax_Request",
            "_ResponseSource": "Broswer",
            "Referer": referer,
            "Origin": self.base_url,
            "Accept": "*/*",
            "Cache-Control": "no-cache",
            "Pragma": "no-cache",
            "__RequestVerificationToken": self.token or "",
            "Cookie": f"SessionID={self.session_id}" if self.session_id else "",
        }

        if self.debug:
            self._dlog("[DEBUG] POST /api/sms/sms-list")
            self._dlog(f"[DEBUG] Headers: {headers}")
            self._dlog(f"[DEBUG] Data: {xml_req}")

        r = self.session.post(url, data=xml_req, headers=headers, timeout=10)
        self.process_http_headers(r)
        r.raise_for_status()

        # Ответ в XML
        root = ET.fromstring(r.content)
        if root.tag == "error":
            code = root.findtext("code")
            raise RuntimeError(f"sms-list error code={code}")

        return self._parse_sms_list(root)

    def delete_sms(self, indexes: list[str]) -> None:
        """Удаление SMS по ID (Index) через /api/sms/delete-sms.

        Поддерживает удаление нескольких сообщений за один вызов.
        """
        if not indexes:
            return

        url = urljoin(self.base_url, "/api/sms/delete-sms")
        referer = urljoin(self.base_url, "/html/smsinbox.html?smsinbox")

        # Вариант 1: повторяющиеся <Index>...</Index>
        indexes_xml = "".join(f"<Index>{i}</Index>" for i in indexes)
        xml_variants = [
            (
                '<?xml version="1.0" encoding="UTF-8"?>'
                "<request>"
                f"{indexes_xml}"
                "</request>"
            ),
            # Вариант 2: <Indexes><Index>..</Index></Indexes>
            (
                '<?xml version="1.0" encoding="UTF-8"?>'
                "<request>"
                "<Indexes>"
                f"{indexes_xml}"
                "</Indexes>"
                "</request>"
            ),
        ]

        headers = {
            "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
            "X-Requested-With": "XMLHttpRequest",
            "X-KL-saas-Ajax-Request": "Ajax_Request",
            "_ResponseSource": "Broswer",
            "Referer": referer,
            "Origin": self.base_url,
            "Accept": "*/*",
            "Cache-Control": "no-cache",
            "Pragma": "no-cache",
            "__RequestVerificationToken": self.token or "",
            "Cookie": f"SessionID={self.session_id}" if self.session_id else "",
        }

        last_err = None
        for xml_req in xml_variants:
            if self.debug:
                self._dlog("[DEBUG] POST /api/sms/delete-sms")
                self._dlog(f"[DEBUG] Headers: {headers}")
                self._dlog(f"[DEBUG] Data: {xml_req}")

            r = self.session.post(url, data=xml_req, headers=headers, timeout=10)
            self.process_http_headers(r)
            if r.status_code != 200:
                last_err = RuntimeError(f"delete-sms http {r.status_code}")
                continue

            # Parse XML from bytes
            try:
                root = ET.fromstring(r.content)
            except ET.ParseError:
                last_err = RuntimeError("delete-sms: invalid XML response")
                continue

            if root.tag == "error":
                code = root.findtext("code")
                last_err = RuntimeError(f"delete-sms error code={code}")
                continue

            # Many firmwares return <response>OK</response>
            if root.tag == "response":
                if (root.text or "").strip().upper() == "OK":
                    return

            # Some return <response><.../></response> without OK text
            return

        raise last_err or RuntimeError("delete-sms failed")

    def _parse_sms_list(self, root: ET.Element) -> dict:
        """Parse /api/sms/sms-list response into dict."""
        # Expected:
        # <response><Count>..</Count><Messages><Message>...</Message></Messages></response>
        out = {
            "count": None,
            "messages": [],
        }
        count = root.findtext("Count")
        if count is not None:
            try:
                out["count"] = int(count)
            except ValueError:
                out["count"] = count

        messages = root.find("Messages")
        if messages is None:
            return out

        for msg in messages.findall("Message"):
            # Common fields: Index, Phone, Content, Date, Smstat, SmsType, Priority, SaveType
            phone = (
                msg.findtext("Phone")
                or msg.findtext("PhoneNum")
                or msg.findtext("PhoneNumber")
                or msg.findtext("Number")
            )
            out["messages"].append(
                {
                    "index": msg.findtext("Index"),
                    "id": msg.findtext("Index"),
                    "from": phone,
                    "phone": phone,
                    "content": msg.findtext("Content"),
                    "date": msg.findtext("Date"),
                    "smstat": msg.findtext("Smstat"),
                    "sms_type": msg.findtext("SmsType"),
                }
            )
        return out


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

    parser = argparse.ArgumentParser(description="Чтение SMS с Huawei B618s-22d (HiLink)")
    parser.add_argument("--url", default=os.getenv("HUAWEI_URL"), help="URL роутера (или env HUAWEI_URL)")
    parser.add_argument("--username", default=os.getenv("HUAWEI_USERNAME"), help="Логин (или env HUAWEI_USERNAME)")
    parser.add_argument("--password", default=os.getenv("HUAWEI_PASSWORD"), help="Пароль (или env HUAWEI_PASSWORD)")
    parser.add_argument("--box", choices=BOX_TYPES.keys(), default="inbox", help="Папка SMS")
    parser.add_argument("--page", type=int, default=1, help="PageIndex (с 1)")
    parser.add_argument("--count", type=int, default=20, help="ReadCount")
    parser.add_argument("--unread-preferred", type=int, choices=[0, 1], default=0, help="UnreadPreferred")
    parser.add_argument("--delete", nargs="+", help="Удалить SMS по ID (Index). Пример: --delete 40042 40038")
    parser.add_argument("--json", action="store_true", help="Вывод JSON (вместо таблицы)")
    parser.add_argument("--debug", action="store_true", help="Подробные логи")
    args = parser.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 = HuaweiHiLinkClient(args.url, args.username, args.password, debug=args.debug)

    if not client.ensure_logged_in():
        safe_print("[ERR] SCRAM login failed")
        return 1

    # Удаление по ID (Index)
    if args.delete:
        try:
            client.delete_sms([str(x) for x in args.delete])
            safe_print(f"[OK] deleted: {', '.join(args.delete)}")
        except Exception as e:
            safe_print(f"[ERR] Failed to delete SMS: {e}")
            return 1

    try:
        data = client.sms_list(
            box_type=BOX_TYPES[args.box],
            page_index=args.page,
            read_count=args.count,
            unread_preferred=args.unread_preferred,
        )
    except Exception as e:
        safe_print(f"[ERR] Failed to read SMS: {e}")
        return 1

    if args.json:
        safe_print(json.dumps(data, ensure_ascii=False, indent=2))
        return 0

    msgs = data.get("messages", [])
    safe_print(f"[OK] box={args.box} page={args.page} count={len(msgs)} total={data.get('count')}")
    for m in msgs:
        content = (m.get("content") or "").replace("\r", " ").replace("\n", " ").strip()
        if len(content) > 120:
            content = content[:117] + "..."
        safe_print(f"- [id={m.get('index')}] [{m.get('date')}] from={m.get('phone')}: {content}")

    return 0


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