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

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

import requests
from urllib.parse import urljoin
import sys
import argparse
import os
import xml.etree.ElementTree as ET
import time
import re
from datetime import datetime

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)


class SMSSender:
    def __init__(self, base_url, username, password, debug=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 process_http_headers(self, response):
        """Обработка заголовков ответа для обновления SessionID и Token"""
        if 'SessionID' in response.cookies:
            self.session_id = response.cookies['SessionID']
            self.session.cookies.set('SessionID', self.session_id)
            if self.debug:
                safe_print(f"[DEBUG] Обновлен SessionID = {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
                if self.debug:
                    safe_print(f"[DEBUG] Обновлен Token = {self.token[:50]}...")
    
    def get_session_token(self, force_refresh=False):
        """Получение SessionID и Token для Huawei API"""
        try:
            url = urljoin(self.base_url, '/api/webserver/SesTokInfo')
            if force_refresh:
                self.session_id = None
                self.token = None
            
            response = self.session.get(url, timeout=10)
            self.process_http_headers(response)
            
            if response.status_code == 200:
                try:
                    root = ET.fromstring(response.text)
                    ses_info = root.find('SesInfo')
                    tok_info = root.find('TokInfo')
                    
                    if ses_info is not None and ses_info.text:
                        self.session_id = ses_info.text
                    if tok_info is not None and tok_info.text:
                        self.token = tok_info.text
                    
                    if self.session_id and self.token:
                        if self.debug and not force_refresh:
                            safe_print("[DEBUG] Получены SessionID и Token")
                        return True
                except ET.ParseError:
                    # Простой парсинг без XML
                    text = response.text
                    if 'SesInfo' in text and 'TokInfo' in text:
                        ses_match = re.search(r'<SesInfo>(.*?)</SesInfo>', text)
                        tok_match = re.search(r'<TokInfo>(.*?)</TokInfo>', text)
                        if ses_match and tok_match:
                            self.session_id = ses_match.group(1)
                            self.token = tok_match.group(1)
                            if self.debug and not force_refresh:
                                safe_print("[DEBUG] Получены SessionID и Token")
                            return True
        except Exception as e:
            if self.debug:
                safe_print(f"[DEBUG] Ошибка при получении токена: {e}")
        
        return False
    
    def login(self):
        """Авторизация для вызовов API.

        Для некоторых прошивок B618 требуется SCRAM (challenge_login/authentication_login).
        """
        if self._scram_login():
            return True

        # Fallback: Basic Auth (может открыть веб-страницы, но не всегда даёт права на SMS API)
        try:
            # Устанавливаем Basic Auth
            self.session.auth = (self.username, self.password)
            
            # Загружаем страницу SMS для получения сессии
            url = urljoin(self.base_url, '/html/smsinbox.html')
            response = self.session.get(url, timeout=10)
            
            if response.status_code == 200:
                if self.debug:
                    safe_print("[DEBUG] Авторизация через Basic Auth")
                
                # Сохраняем все Cookie
                if response.cookies:
                    for cookie in response.cookies:
                        self.session.cookies.set(cookie.name, cookie.value)
                
                # Получаем токен через API
                time.sleep(0.3)
                self.get_session_token(force_refresh=True)
                return True
        except Exception as e:
            if self.debug:
                safe_print(f"[DEBUG] Ошибка при авторизации: {e}")
        
        return False

    def _scram_login(self) -> bool:
        """SCRAM логин (challenge_login/authentication_login)."""
        try:
            # 1) setup session
            r = self.session.get(urljoin(self.base_url, "/"), timeout=10)
            self.process_http_headers(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)
            if r.status_code != 200:
                return False
            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 = (
                f'<?xml version="1.0" encoding="UTF-8"?>'
                f"<request>"
                f"<username>{self.username}</username>"
                f"<firstnonce>{clientnonce}</firstnonce>"
                f"<mode>1</mode>"
                f"</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)
            if r.status_code != 200:
                return False

            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

            # token for next step comes from response header
            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 = (
                f'<?xml version="1.0" encoding="UTF-8"?>'
                f"<request>"
                f"<clientproof>{clientproof}</clientproof>"
                f"<finalnonce>{servernonce}</finalnonce>"
                f"</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)
            if r.status_code != 200:
                return False

            # Токен иногда приходит в __RequestVerificationTokenone
            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
                if self.debug:
                    safe_print(f"[DEBUG] SCRAM token = {tok[:50]}...")

            # Refresh SesTokInfo after SCRAM
            self.get_session_token(force_refresh=True)
            if self.session_id and self.token:
                if self.debug:
                    safe_print("[DEBUG] Авторизация через SCRAM")
                return True

        except Exception as e:
            if self.debug:
                safe_print(f"[DEBUG] SCRAM login failed: {e}")

        return False
    
    def send_sms(self, phone_number, message):
        """Отправка SMS через Huawei B618s-22d API"""
        if not self.login():
            safe_print("[ERR] Не удалось авторизоваться")
            return False
        
        # Убеждаемся, что у нас есть токен и сессия
        if not self.session_id or not self.token:
            safe_print("[WARN] Получаем токен перед отправкой SMS...")
            self.get_session_token(force_refresh=True)
        
        if not self.session_id or not self.token:
            safe_print("[ERR] Не удалось получить токен для отправки SMS")
            return False
        
        return self._send_sms_via_api(phone_number, message)
    
    def _send_sms_via_api(self, phone_number, message):
        """Отправка SMS через /api/sms/send-sms."""
        endpoint = '/api/sms/send-sms'
        
        # Номер телефона с плюсом (как в реальном запросе)
        clean_phone = phone_number.strip()
        if not clean_phone.startswith('+'):
            if len(clean_phone.replace('-', '').replace(' ', '').replace('(', '').replace(')', '')) > 10:
                clean_phone = '+' + clean_phone
        
        # Текущая дата и время
        current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        # Формат XML как в веб-интерфейсе
        xml_data = (
            f'<?xml version="1.0" encoding="UTF-8"?>'
            f'<request>'
            f'<Index>-1</Index>'
            f'<Phones><Phone>{clean_phone}</Phone></Phones>'
            f'<Sca></Sca>'
            f'<Content>{message}</Content>'
            f'<Length>{len(message)}</Length>'
            f'<Reserved>0</Reserved>'
            f'<Date>{current_date}</Date>'
            f'</request>'
        )
        
        url = urljoin(self.base_url, endpoint)
        sms_page_url = urljoin(self.base_url, '/html/smsinbox.html?smsinbox')
        
        # Заголовки как в веб-интерфейсе
        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': sms_page_url,
            'Origin': self.base_url,
            'Accept': '*/*',
            'Accept-Language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7',
            'Cache-Control': 'no-cache',
            'Pragma': 'no-cache',
            'Connection': 'keep-alive',
            '__RequestVerificationToken': self.token,
            'Cookie': f'SessionID={self.session_id}'
        }
        
        # Устанавливаем Cookie в сессию
        self.session.cookies.set('SessionID', self.session_id)
        
        try:
            if self.debug:
                safe_print(f"[DEBUG] Отправка SMS на {endpoint}")
                safe_print(f"[DEBUG] Headers: {headers}")
                safe_print(f"[DEBUG] Data: {xml_data[:200]}...")
            
            response = self.session.post(url, data=xml_data, headers=headers, timeout=10)
            self.process_http_headers(response)
            
            if self.debug:
                safe_print(f"[DEBUG] Status: {response.status_code}")
                safe_print(f"[DEBUG] Response: {response.text[:300]}")
            
            if response.status_code == 200:
                response_text = response.text
                response_lower = response_text.lower()
                
                # Проверяем на ошибки
                if 'error' in response_lower and 'code' in response_lower:
                    try:
                        root = ET.fromstring(response_text)
                        error_code = root.find('.//code')
                        if error_code is not None:
                            error_code_text = error_code.text
                            safe_print(f"[ERR] Ошибка отправки SMS: код {error_code_text}")
                            if error_code_text == '100003':
                                safe_print("  Проблема с авторизацией. Проверьте логин и пароль.")
                            return False
                    except:
                        pass
                    return False
                
                # Проверяем на успех
                if any(word in response_lower for word in ['ok', 'success', 'sent']):
                    safe_print(f"[OK] SMS отправлена на {phone_number}")
                    return True
                
                # Проверяем XML ответ
                try:
                    root = ET.fromstring(response_text)
                    if root.find('response') is not None and root.find('error') is None:
                        safe_print(f"[OK] SMS отправлена на {phone_number}")
                        return True
                    if root.find('.//Phone') is not None:
                        safe_print(f"[OK] SMS отправлена на {phone_number}")
                        return True
                except ET.ParseError:
                    if len(response_text) < 200 and 'error' not in response_lower:
                        safe_print(f"[OK] SMS отправлена на {phone_number}")
                        return True
                except:
                    pass
        
        except Exception as e:
            if self.debug:
                safe_print(f"[DEBUG] Ошибка при отправке: {e}")
            safe_print(f"[ERR] Ошибка при отправке SMS: {e}")
        
        return False


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

    parser = argparse.ArgumentParser(description='Отправка SMS через веб-интерфейс Huawei B618s-22d')
    parser.add_argument('phone', help='Номер телефона получателя')
    parser.add_argument('message', help='Текст сообщения')
    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('--debug', action='store_true',
                       help='Включить режим отладки (показывать детали запросов)')
    
    args = parser.parse_args()

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

    sender = SMSSender(args.url, args.username, args.password, debug=args.debug)
    success = sender.send_sms(args.phone, args.message)
    
    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()