ZteModemClient

ZteModemClient.py

# hint: developed for and tested with ZTE ZTE MF79U (version BD_XCBZHKMF79UV1.0.0B03)

import requests
import hashlib
import re
import logging
from typing import Dict, Any, Tuple

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class ZteModemClient:
    HEADERS = {
        "Accept": "application/json, text/javascript, */*; q=0.01"
    }

    def __init__(self, protocol: str, host: str, port: str, password: str):
        self.url = f"{protocol}://{host}:{port}"
        self.password = password
        self.cookie = ''
        self.HEADERS["Host"] = f"{host}:{port}"
        self.HEADERS["Referer"] = f"{self.url}/index.html"

    def _get_request(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
        response = requests.get(f"{self.url}/{endpoint}", params=params, headers=self.HEADERS)
        response.raise_for_status()
        return response.json(), response

    def _post_request(self, endpoint: str, data: Dict[str, Any]) -> Tuple[Dict[str, Any], requests.Response]:
        headers = self.HEADERS.copy()

        if self.cookie:
            headers["Cookie"] = self.cookie

        response = requests.post(f"{self.url}/{endpoint}", data=data, headers=headers)
        response.raise_for_status()
        return response.json(), response

    def get_device_version(self) -> Dict[str, str]:
        params = {
            "isTest": "false",
            "cmd": "Language,cr_version,wa_inner_version",
            "multi_data": "1"
        }
        return self._get_request("goform/goform_get_cmd_process", params)[0]


    def login(self) -> None:
        logger.info("Starting login process...")

        password_hash = self.get_password_hash()

        params = {
            "isTest": "false",
            "goformId": "LOGIN",
            "password": password_hash
        }

        response_json, response = self._post_request("goform/goform_set_cmd_process", params)

        if response_json['result'] != '0':
            raise Exception(f"Login failed. Result: {response_json['result']}")

        cookie_header = response.headers.get("Set-Cookie")
        if not cookie_header:
            raise Exception("Login failed. No Set-Cookie header found.")

        cookie_match = re.search(r'stok=([A-F0-9]+)', cookie_header)

        if cookie_match:
            self.cookie = cookie_match.group(1)
        else:
            raise Exception("Login failed. Cookie not found in Set-Cookie header.")

        logger.info("Login successful!")
        logger.debug(f"Cookie: {self.cookie}")

        return response_json

    def get_ad(self) -> str:
        device_info = self.get_device_version()
        cr_version = device_info['cr_version']
        wa_inner_version = device_info['wa_inner_version']
        logger.debug(f"Device info: CR Version: {cr_version}, WA Inner Version: {wa_inner_version}")

        rd = self.get_rd()
        logger.debug(f"RD: {rd}")

        prefix_hash = hashlib.md5((cr_version + wa_inner_version).encode('utf-8')).hexdigest()
        ad = hashlib.md5((prefix_hash + rd).encode('utf-8')).hexdigest()
        logger.debug(f"AD: {ad}")
        return ad

    def get_ld(self) -> str:
        params = {"isTest": "false", "cmd": "LD"}
        return self._get_request("goform/goform_get_cmd_process", params)[0]['LD']

    def get_rd(self) -> str:
        params = {"isTest": "false", "cmd": "RD"}
        return self._get_request("goform/goform_get_cmd_process", params)[0]['RD']

    def get_password_hash(self) -> str:
        ld = self.get_ld()
        logger.debug(f"LD: {ld}")

        prefix_hash = hashlib.sha256(self.password.encode('utf-8')).hexdigest().upper()
        return hashlib.sha256((prefix_hash + ld.upper()).encode('utf-8')).hexdigest().upper()

    def send_cmd(self, cmd: str, not_callback: bool = False) -> Dict[str, Any]:
        params = {
            "isTest": "false",
            "goformId": cmd
        }

        if not_callback:
            params["notCallback"] = "true"

        ad = self.get_ad()
        if ad:
            params["AD"] = ad

        response, _ = self._post_request("goform/goform_set_cmd_process", params)
        return response

    def get_cmd(self, cmd: str, not_callback: bool = False) -> Dict[str, Any]:
        params = {
            "isTest": "false",
            "multi_data": "1",
            "cmd": cmd
        }

        if not_callback:
            params["notCallback"] = "true"

        return self._get_request("goform/goform_get_cmd_process", params)

def main():
    # Connection details
    protocol = 'http'
    host = '192.168.0.1'
    port = '80'
    password = ''

    zte_client = ZteModemClient(protocol, host, port, password)

    try:
        response = zte_client.login()
        print(response.get('result')) # 0

        response = zte_client.send_cmd(cmd="DISCONNECT_NETWORK")
        print(response.get('result')) # success

        response = zte_client.send_cmd(cmd="CONNECT_NETWORK")
        print(response.get('result')) # success
    except Exception as e:
        logger.error(f"An error occurred: {str(e)}")

if __name__ == "__main__":
    main()