automation – Malware University

The following Python code may be used to remotely control a Virtualbox-managed virtual machine. It is useful for batch testing or malware analysis (sandbox management).

#!/usr/bin/env python3

#
# Installing the virtualbox deps:
#   python3 -m pip install pyvbox
#   Download the VirtualBox SDK from https://www.virtualbox.org/wiki/Downloads
#   Unzip the package
#   cd sdk/installer
#   python3 vboxapisetup.py install
#

import virtualbox
import logging


class Sandbox(object):
    def __init__(self, name):
        vbox = virtualbox.VirtualBox()
        self.vm = vbox.find_machine(name)
        #self.session = virtualbox.Session()
        self.session = virtualbox.library.ISession(self.vm.create_session())

        return


    def start(self):
        logging.info('Starting the VM from Office snapshot')
        self.snapshot = self.vm.find_snapshot('Office 2007 Live')
        progress = self.session.machine.restore_snapshot(self.snapshot)
        progress.wait_for_completion()
        self.session.unlock_machine()
        # progress = self.vm.launch_vm_process(self.session, 'headless', '')
        progress = self.vm.launch_vm_process(self.session, '', '')
        progress.wait_for_completion()

        if self.vm.state == self.vm.state.running:
            print('Successfully running')
        print('Done')

        # Create guest session.
        self.console = virtualbox.library.IConsole(self.session.console)
        self.guest = self.console.guest
        # self.guest.sessions[0] contains our IGuestSession object
        self.os_type = self.guest.os_type_id	    # Example
        # self.guest = virtualbox.library.IGuest(self.session)
        # Session is already established in snapshot of unlocked machine?
        self.g_session = self.guest.create_session('malware', 'password', '', '')

        return


    #
    # Username is 'malware'; password is 'password'
    #
    def guest_session_test(self):
        self.console = virtualbox.library.IConsole(self.session.console)
        self.guest = self.console.guest
        # self.guest.sessions[0] contains our IGuestSession object
        self.os_type = self.guest.os_type_id	    # Example
        # self.guest = virtualbox.library.IGuest(self.session)
        # Session is already established in snapshot of unlocked machine?
        self.g_session = self.guest.create_session('malware', 'password', '', '')
        # 1 = R
        # 2 = 
        # 3 = RW
        flag = virtualbox.library.FileCopyFlag(0)
        # This stores the file in your user's home directory by default.
        self.g_session.file_copy_from_guest('C:/Program Files (x86)/Microsoft Office/Office12/GRAPH.EXE', 'C:/WINDOWS/Temp/graph.exe', [flag])


    def drop_artifact(self, artifact_path, destination_on_guest):
        logging.info('About to drop "{}" to "{}"'.format(artifact_path, destination_on_guest))
        flag = virtualbox.library.FileCopyFlag(0)
        progress = self.g_session.file_copy_to_guest(artifact_path, destination_on_guest, [flag])
        progress.wait_for_completion()
        logging.info('"{}" successfully dropped to guest'.format(artifact_path))

        return


    def receive_artifact(self, src_on_guest, path_host_save):
        logging.info('About to pull "{}" to "{}"'.format(src_on_guest, path_host_save))
        flag = virtualbox.library.FileCopyFlag(0)
        progress = self.g_session.file_copy_from_guest(src_on_guest, path_host_save, [flag])
        progress.wait_for_completion()
        logging.info('"{}" successfully saved to "{}"'.format(src_on_guest, path_host_save))

        return


    def execute_command(self, 
                        executable_path: str, 
                        arguments: [str],
                        environment_changes: [str],
                        process_create_flag: [virtualbox.library.ProcessCreateFlag],
                        timeout_ms: int,
                        priority: virtualbox.library.ProcessPriority,
                        affinity: [int]):
        terminate_flag = virtualbox.library.ProcessWaitForFlag(2)

        self.guest_process = self.g_session.process_create_ex(  executable_path, 
                                                                arguments, 
                                                                environment_changes,
                                                                process_create_flag,
                                                                timeout_ms,
                                                                priority,
                                                                affinity)
        logging.info('Process "{}" successfully kicked off...waiting for it to finish'.format(executable_path))
        self.process_wait_result = self.guest_process.wait_for(terminate_flag)
        if self.process_wait_result == virtualbox.library.ProcessWaitResult(5):
            logging.info('Process "{}" exceeded the timeout threashold of {} milliseconds'.format(  executable_path,
                                                                                                    str(timeout_ms))) 
        elif self.process_wait_result == virtualbox.library.ProcessWaitResult(2):
            logging.info('Process "{}" terminated gracefully'.format(executable_path))


    def stop(self):
        # Shut off VM
        progress = self.session.console.power_down()
        progress.wait_for_completion()

        return


s = Sandbox('Windows 7 Office')
s.start()
# s.guest_session_test()
s.drop_artifact('C:/WINDOWS/System32/cmd.exe', 'C:/WINDOWS/Temp/cmd_2016.exe')
#s.execute_command(  'C:/WINDOWS/Temp/cmd_2016.exe',
s.execute_command(  'C:/Program Files (x86)/Microsoft Office/Office12/GRAPH.EXE',
                    [],
                    [],
                    [virtualbox.library.ProcessCreateFlag(0)],
                    20000,
                    virtualbox.library.ProcessPriority(1),
                    [])
s.stop()

One of the more popular Cobalt Strike attacks involves building an Office VBA payload. Unsophisticated attackers will not likely modify the generated payloads from the framework. This gives a defender an easier opportunity to find CS payloads and easily dump/analyze them.

The following tool (Python3) will help an analyst or security system to discover functionality of the embedded payload. Once extracted, you can use an online disassembler like ODA to analyze the x86 shellcode. Keep in mind it has both code and data embedded. Hence we leave possible decoded ASCII in a separate view.

#!/usr/bin/env python3
# (@) - Extracts shellcode from Array() in VBA Macro attack
#       Also prints possible ASCII values embedded in payload
#
import sys
import re

content_file=sys.argv[1]
contents = ''

try:
    with open(content_file, 'r') as f:
        contents = f.read()
except Exception as e:
    print(str(e))
    print('Usage:  python3 ' + sys.argv[0] + ' <generated_vba_file>')
    raise

start_location = contents.find('myArray = Array(')
end_location = contents.index('If Len(Environ("ProgramW6432"')

what_we_want = contents[start_location:end_location - 1]
what_we_want = re.sub(r'^.*?\(', ' ', what_we_want)
what_we_want = what_we_want.replace(' _', '')
what_we_want = what_we_want.replace(")", "")
what_we_want = what_we_want.replace(',', ' ')

disasm_this = ''
text_this = ''
for element in what_we_want.split(' '):
    if not element:
        continue
    new_number = int(element)
    asm_num = format(new_number & 0xff, "02X")
    disasm_this += asm_num + ' '
    if new_number < 0:
        continue
    text_this += bytes.fromhex(asm_num).decode('ascii')
print('x86 payload:')
print(disasm_this)
print('\n\n')
print('Possible ASCII data found:')
print(text_this)

Tag: automation

Virtualbox Automation

Cobalt Strike MS Office Macro Shellcode Decoder/Dumper