zte-exporter/zte_exporter.py

#!/bin/python3

import requests
from requests.exceptions import RequestException
import hashlib
from datetime import datetime
import urllib.parse
import json
import sys
import time
import urllib3
from http.server import BaseHTTPRequestHandler, HTTPServer
import argparse

parser = argparse.ArgumentParser("ZTE simple exporter")
parser.add_argument("port", help="Serve metrics on what port", type=int)
parser.add_argument("ip", help="Router ip address")
parser.add_argument("username", help="Router username", default="admin")
parser.add_argument("password", help="Router password")
parser.add_argument("--login", help="Login method (multi, single)", default="single")
args = parser.parse_args()

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

s = requests.Session()

class zteRouter:

    def __init__(self, ip, username, password, login):
        self.login = login
        self.ip = ip
        self.protocol = "http"  # default to http
        self.username = username
        self.password = password
        self.try_set_protocol()
        self.referer = f"{self.protocol}://{self.ip}/"

    def try_set_protocol(self):
        protocols = ["http", "https"]
        for protocol in protocols:
            url = f"{protocol}://{self.ip}"
            try:
                response = requests.get(url, timeout=5, verify=False)
                if response.ok:
                    self.protocol = protocol
                    # print(f"{self.ip} is accessible via {protocol}")
                    return
            except RequestException:
                pass  # If RequestException occurs, try the next protocol
        # print(f"Could not determine the protocol for {self.ip}")

    def hash(self, str):
        return hashlib.sha256(str.encode()).hexdigest()

    def get_LD(self):
        header = {"Referer": self.referer}
        payload = "isTest=false&cmd=LD"
        r = s.get(self.referer + f"goform/goform_get_cmd_process?{payload}&_=", headers=header, data=payload, verify=False)
        return r.json()["LD"].upper()

    def getCookie(self, username, password, LD, login):
        header = {"Referer": self.referer}
        hashPassword = self.hash(password).upper()
        ztePass = self.hash(hashPassword + LD).upper()

        if login == "multi":
            payload = {
                'isTest': 'false',
                'goformId': 'LOGIN_MULTI_USER',
                'password': ztePass,
                'user': username
            }
        else:
            payload = {
                'isTest': 'false',
                'goformId': 'LOGIN',
                'password': ztePass,
            }            

        r = s.post(self.referer + "goform/goform_set_cmd_process", headers=header, data=payload, verify=False)
        return "stok=" + r.cookies["stok"].strip('\"')

    def zteinfo(self):
        ip = self.ip
        cookie = self.getCookie(username=self.username, password=self.password, LD=self.get_LD(), login=self.login)
        cmd_url = f"{self.protocol}://{self.ip}/goform/goform_get_cmd_process?multi_data=1&isTest=false&cmd=wa_inner_version%2Clte_rssi%2CZ5g_rssi%2Clte_rsrp%2CZ5g_rsrp%2Clte_rsrq%2CZ5g_rsrq%2Clte_snr%2CZ5g_SINR%2Csignalbar%2Ccell_id%2Cnr5g_action_band%2Ccell_id%2Cnetwork_provider%2Cmonthly_tx_bytes%2Cmonthly_rx_bytes%2Crealtime_tx_bytes%2Crealtime_rx_bytes%2Crealtime_tx_thrpt%2Crealtime_rx_thrpt%2Cwan_ipaddr%2Cwan_apn%2Clte_multi_ca_scell_sig_info%2Cnr_multi_ca_scell_info%2Cnr5g_action_band"
        cookie_pass = cookie

        headers = {
            "Host": ip,
            "Referer": f"{self.referer}index.html",
            "Cookie": f"{cookie_pass}"
        }

        response = s.get(cmd_url, headers=headers, verify=False)
        return response.text
    
    def iterateCellInfo(self, str, type):
        for f in str:
                print (f)
    

zteInstance = zteRouter(args.ip, args.username, args.password, args.login)
serverPort = args.port

class serveInfos(BaseHTTPRequestHandler):
    def do_GET(self):
        gatheredJson = json.loads(zteInstance.zteinfo())
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.end_headers()

        #self.wfile.write(bytes("\n\n%s\n\n" % json.dumps(gatheredJson, indent=4), "utf-8"))

        # for reference for now
        self.wfile.write(bytes("# HELP zte_modem_info Modem general information\n# TYPE zte_modem_info untyped\n", "utf-8"))
        self.wfile.write(bytes("zte_modem_info{firmware=\"%s\",ip_addr=\"" % gatheredJson["wa_inner_version"], "utf-8"))
        self.wfile.write(bytes("%s\",apn=\"" % gatheredJson["wan_ipaddr"], "utf-8"))
        self.wfile.write(bytes("%s\",cell_id=\"" % gatheredJson["wan_apn"], "utf-8"))
        self.wfile.write(bytes("%s\"} 1\n\n" % gatheredJson["cell_id"], "utf-8"))

        self.wfile.write(bytes("# HELP zte_rsrp Reference Signal Received Power\n# TYPE zte_rsrp gauge\n", "utf-8"))
        self.wfile.write(bytes("# HELP zte_snr Signal-to-Interference-plus-Noise Ratio\n# TYPE zte_snr gauge\n", "utf-8"))
        self.wfile.write(bytes("# HELP zte_rsrq Reference Signal Received Quality\n# TYPE zte_rsrq gauge\n", "utf-8"))
        self.wfile.write(bytes("# HELP zte_rssi Received Signal Strength Indicator\n# TYPE zte_rssi gauge\n", "utf-8"))

        if gatheredJson['nr_multi_ca_scell_info'] and gatheredJson['nr5g_action_band']:
            self.wfile.write(bytes("zte_rsrq{band=\"%s\"} " % gatheredJson["nr5g_action_band"], "utf-8"))
            self.wfile.write(bytes("%s\n" % gatheredJson["Z5g_rsrq"], "utf-8"))

            self.wfile.write(bytes("zte_rsrp{band=\"%s\"} " % gatheredJson["nr5g_action_band"], "utf-8"))
            self.wfile.write(bytes("%s\n" % gatheredJson["Z5g_rsrp"], "utf-8"))
                                                      
            self.wfile.write(bytes("zte_rssi{band=\"%s\"} " % gatheredJson["nr5g_action_band"], "utf-8"))
            self.wfile.write(bytes("%s\n" % gatheredJson["Z5g_rssi"], "utf-8"))

            self.wfile.write(bytes("zte_snr{band=\"%s\"} " % gatheredJson["nr5g_action_band"], "utf-8"))
            self.wfile.write(bytes("%s\n\n" % gatheredJson["Z5g_SINR"], "utf-8"))
        
            for index, item in enumerate(gatheredJson['nr_multi_ca_scell_info'].split(";")):
                nrbit = item.split(",")
                if not len(nrbit[0]) == 0:
                    self.wfile.write(bytes("zte_rsrq{band=\"%s\"} " % nrbit[3], "utf-8"))
                    self.wfile.write(bytes("%s\n" % nrbit[8], "utf-8"))

                    self.wfile.write(bytes("zte_rsrp{band=\"%s\"} " % nrbit[3], "utf-8"))
                    self.wfile.write(bytes("%s\n" % nrbit[7], "utf-8"))

                    self.wfile.write(bytes("zte_rssi{band=\"%s\"} " % nrbit[3], "utf-8"))
                    self.wfile.write(bytes("%s\n" % nrbit[10], "utf-8"))

                    self.wfile.write(bytes("zte_snr{band=\"%s\"} " % nrbit[3], "utf-8"))
                    self.wfile.write(bytes("%s\n\n" % nrbit[9], "utf-8"))

        else:
            if gatheredJson["lte_rsrq"]:
                self.wfile.write(bytes("zte_rsrq{band=\"lte\"} %s\n" % gatheredJson["lte_rsrq"], "utf-8"))
            if gatheredJson["Z5g_rsrq"]:
                self.wfile.write(bytes("zte_rsrq{band=\"5g\"} %s\n" % gatheredJson["Z5g_rsrq"], "utf-8"))
            if gatheredJson["lte_rsrp"]:
                self.wfile.write(bytes("zte_rsrp{band=\"lte\"} %s\n" % gatheredJson["lte_rsrp"], "utf-8"))
            if gatheredJson["Z5g_rssi"]:
                self.wfile.write(bytes("zte_rsrp{band=\"5g\"} %s\n" % gatheredJson["Z5g_rssi"], "utf-8"))
            if gatheredJson["lte_rssi"]:
                self.wfile.write(bytes("zte_snr{band=\"lte\"} %s\n" % gatheredJson["lte_rssi"], "utf-8"))
            if gatheredJson["Z5g_SINR"]:
                self.wfile.write(bytes("zte_snr{band=\"5g\"} %s\n" % gatheredJson["Z5g_SINR"], "utf-8"))
            if gatheredJson["lte_snr"]:
                self.wfile.write(bytes("zte_snr{band=\"lte\"} %s\n" % gatheredJson["lte_snr"], "utf-8"))
            if gatheredJson["Z5g_SINR"]:
                self.wfile.write(bytes("zte_snr{band=\"5g\"} %s\n\n" % gatheredJson["Z5g_SINR"], "utf-8"))

        self.wfile.write(bytes("# HELP zte_bandwidth_used_monthly_tx Used bandwidth, transmit (bytes)\n# TYPE zte_bandwidth_used_monthly_tx counter\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_used_monthly_tx %s\n\n" % gatheredJson["monthly_tx_bytes"], "utf-8"))
        self.wfile.write(bytes("# HELP zte_bandwidth_used_monthly_rx Used bandwidth, receive (bytes)\n# TYPE zte_bandwidth_used_monthly_rx counter\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_used_monthly_rx %s\n\n" % gatheredJson["monthly_rx_bytes"], "utf-8"))

        self.wfile.write(bytes("# HELP zte_bandwidth_used_session_tx Used bandwidth, current session, transmit (bytes)\n# TYPE zte_bandwidth_used_session_tx counter\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_used_session_tx %s\n\n" % gatheredJson["realtime_tx_bytes"], "utf-8"))
        self.wfile.write(bytes("# HELP zte_bandwidth_used_session_rx Used bandwidth, current session, receive (bytes)\n# TYPE zte_bandwidth_used_session_rx counter\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_used_session_rx %s\n\n" % gatheredJson["realtime_tx_bytes"], "utf-8"))

        self.wfile.write(bytes("# HELP zte_bandwidth_tx Current bandwidth, transmit (bytes)\n# TYPE zte_bandwidth_tx gauge\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_tx %s\n\n" % gatheredJson["realtime_tx_thrpt"], "utf-8"))
        self.wfile.write(bytes("# HELP bandwidth_rx Current bandwidth, receive (bytes)\n# TYPE zte_bandwidth_rx gauge\n", "utf-8"))
        self.wfile.write(bytes("zte_bandwidth_rx %s\n\n" % gatheredJson["realtime_rx_thrpt"], "utf-8"))

if __name__ == "__main__":
    webServer = HTTPServer(("0.0.0.0", serverPort), serveInfos)
    print("Server started http://%s:%s" % ("0.0.0.0", serverPort))
    webServer.serve_forever()

    webServer.server_close()
    print("Server stopped.")