etz69/irhelper

View on GitHub
vol_plugins/hollowfind.py

Summary

Maintainability
F
6 days
Test Coverage
# Author: Monnappa K A
# Email : monnappa22@gmail.com
# Twitter: @monnappa22
# Description: Volatility plugin to detect different types of Process Hollowing 

import os
import volatility.obj as obj
import volatility.utils as utils
from volatility.plugins.taskmods import PSList
import volatility.plugins.vadinfo as vadinfo
import volatility.plugins.malware.malfind as malfind
from volatility.renderers.basic import Address,Hex


hollow_types = dict(enumerate(["Invalid EXE Memory Protection and Process Path Discrepancy",
                             "No VAD Entry For Process Executable",
                             "Process Base Address and Memory Protection Discrepancy"]))

class HollowFind(vadinfo.VADDump):
    """Detects different types of Process Hollowing"""
    
    def __init__(self, config, *args, **kwargs):
        vadinfo.VADDump.__init__(self, config, *args, **kwargs)
        config.remove_option("BASE") 
    
    def update_proc_peb_info(self, psdata):
        self.proc_peb_info = {}
        # Builds a dictionary of process executable information from PEB
        for proc in psdata:
            pid = int(proc.UniqueProcessId)
            self.proc_peb_info[pid] = [proc, 
                                       pid, 
                                       proc.ImageFileName, 
                                       int(proc.InheritedFromUniqueProcessId),
                                       str(proc.CreateTime)]
            if proc.Peb: 
                # gets process information for the process executable from PEB and updates the dictionary 
                mods = proc.get_load_modules()
                for mod in mods:
                    ext = os.path.splitext(str(mod.FullDllName))[1].lower()
                    if (ext == ".exe"):
                        proc_cmd_line = proc.Peb.ProcessParameters.CommandLine
                        proc_image_baseaddr = proc.Peb.ImageBaseAddress
                        mod_baseaddr = mod.DllBase
                        mod_size = mod.SizeOfImage
                        mod_basename = mod.BaseDllName
                        mod_fullname = mod.FullDllName
                        break
                        
                self.proc_peb_info[pid].extend([str(proc_cmd_line), 
                                                Address(proc_image_baseaddr), 
                                                Address(mod_baseaddr), 
                                                Hex(mod_size), 
                                                str(mod_basename),
                                                str(mod_fullname or "")])
                        
            else:
                self.proc_peb_info[pid].extend(["No PEB", Address(0), Address(0), Hex(0), "No PEB", "No PEB"])
                
                
    def update_proc_vad_info(self, proc_peb_info):
        """Builds a dictionary of process executable information from VAD"""
        self.proc_vad_info = {}
        for pid in proc_peb_info:
            self.proc_vad_info[pid] = []
            proc = proc_peb_info[pid][0]
            
            if proc.Peb:
                # gets process information for the process executable from VAD and updates the dictionary
                for vad, addr_space in proc.get_vads(vad_filter = proc._mapped_file_filter):
                    ext = ""
                    vad_found = False
                    if obj.Object("_IMAGE_DOS_HEADER", offset = vad.Start, vm = addr_space).e_magic != 0x5A4D:
                        continue
                    
                    if str(vad.FileObject.FileName or ''):
                        ext = os.path.splitext(str(vad.FileObject.FileName))[1].lower()
                    
                    if (ext == ".exe") or (vad.Start == proc.Peb.ImageBaseAddress):
                        vad_filename =  vad.FileObject.FileName
                        vad_baseaddr = vad.Start
                        vad_size = vad.End - vad.Start
                        vad_protection = vadinfo.PROTECT_FLAGS.get(vad.VadFlags.Protection.v())
                        vad_tag = vad.Tag
                        self.proc_vad_info[pid].extend([str(vad_filename or ''), 
                                                        Address(vad_baseaddr),
                                                        Hex(vad_size), 
                                                        str(vad_protection or ''), 
                                                        str(vad_tag or '')])
                        vad_found = True
                        break
                
                if vad_found == False:
                    self.proc_vad_info[pid].extend(["NA", Address(0), Hex(0), "NA", "NA"])
                          
            else:
                self.proc_vad_info[pid].extend(["No VAD", Address(0), Hex(0), "No VAD", "No VAD"])
    
    def get_proc_peb_info(self):
        return self.proc_peb_info
    
    def get_proc_vad_info(self):
        return self.proc_vad_info
    
    def detect_proc_hollow(self):
        """Detects hollowed processes and returns dictionary with pid as the key and type of process hollowing as value"""
        proc_peb_info = self.get_proc_peb_info()
        proc_vad_info = self.get_proc_vad_info()
        hol_type = None
        self.hollowed_procs = {}
        for pid in proc_peb_info:
            (proc, pid, proc_name, ppid, create_time, proc_cmd_line, proc_image_baseaddr, mod_baseaddr, 
             mod_size, mod_basename, mod_fullname) = proc_peb_info[pid] 
            (vad_filename, vad_baseaddr, vad_size, vad_protection, vad_tag) = proc_vad_info[pid]
            
            if vad_protection == "PAGE_EXECUTE_READWRITE":
                hol_type = 0
                self.hollowed_procs[pid] = hol_type
                
            elif vad_protection == "NA":
                hol_type = 1
                self.hollowed_procs[pid] = hol_type
            
            elif (vad_protection == "PAGE_EXECUTE_WRITECOPY") and (vad_baseaddr != proc_image_baseaddr):
                hol_type = 2
                self.hollowed_procs[pid] = hol_type
        return self.hollowed_procs
    
    
    def update_parent_proc_info(self, proc_peb_info):
        """Builds a dictionary containing parent process information for all the processes"""
        self.parent_proc_info = {}
        for pid in proc_peb_info:
            self.parent_proc_info[pid] = []
            if pid == 4:
                self.parent_proc_info[pid].extend(["", 0])
            else: 
                ppid = int(proc_peb_info[pid][3])
                if ppid in proc_peb_info:
                    ppname = str(proc_peb_info[ppid][2])
                else:
                    ppname = "NA"
                    
                self.parent_proc_info[pid].extend([ppname, ppid])                 
    
    def get_parent_proc_info(self):
        return self.parent_proc_info
    
    def get_similar_procs(self, procid):
        """Given a process id returns a list containing information of similar processes"""
        self.similar_procs = []
        proc_peb_info = self.get_proc_peb_info()
        parent_proc_info = self.get_parent_proc_info()
        pname = proc_peb_info[procid][2]
        create_time = proc_peb_info[procid][4]
        ppname, ppid = parent_proc_info[procid]
        self.similar_procs.append([pname, procid, ppname, ppid, create_time])
        
        for pid in proc_peb_info:
            if pid == procid:
                continue
            if pname == proc_peb_info[pid][2]:
                proc_name = proc_peb_info[pid][2]
                creation_time = proc_peb_info[pid][4]
                parent_name, parent_id = parent_proc_info[pid]
                self.similar_procs.append([proc_name, pid, parent_name, parent_id, creation_time])
        return self.similar_procs
    
                     
    def calculate(self):
        if self._config.PID:
            filter_pid = self._config.PID
            # This is so that when -p option is given it can still enumerate all processes to determine similar processes
            self._config.PID = None
        else:
            filter_pid = None
        ps = PSList(self._config)
        psdata = ps.calculate()
        self.update_proc_peb_info(psdata)
        proc_peb_info = self.get_proc_peb_info()
        self.update_parent_proc_info(proc_peb_info)
        parent_proc_info = self.get_parent_proc_info()
        self.update_proc_vad_info(proc_peb_info)
        hol_procs = self.detect_proc_hollow()
        proc_vad_info = self.get_proc_vad_info()  
        if hol_procs:
            for (hol_pid, hol_type) in hol_procs.items():
                similar_procs = self.get_similar_procs(hol_pid)
                if not filter_pid:
                    yield (proc_peb_info[hol_pid], 
                           proc_vad_info[hol_pid],
                           hol_pid, hol_type,
                           similar_procs,
                           parent_proc_info[hol_pid] )
                else:
                    for p in filter_pid.split(','):
                        fil_pid = int(p)
                        if int(fil_pid) == hol_pid:
                            yield (proc_peb_info[hol_pid],
                                   proc_vad_info[hol_pid],
                                   hol_pid, hol_type,
                                   similar_procs,
                                   parent_proc_info[hol_pid])
    
    def render_text(self, outfd, data):
        for (hol_proc_peb_info, hol_proc_vad_info, hol_pid, hol_type, similar_procs, parent_proc_info) in data:
            (proc, pid, proc_name, ppid, create_time, proc_cmd_line, proc_image_baseaddr, mod_baseaddr, 
             mod_size, mod_basename, mod_fullname) = hol_proc_peb_info
            (vad_filename, vad_baseaddr, vad_size, vad_protection, vad_tag) = hol_proc_vad_info
            (parent_name, parent_id) = parent_proc_info
            
            outfd.write("Hollowed Process Information:\n")
            outfd.write("\tProcess: {0} PID: {1}\n".format(proc_name, hol_pid))
            outfd.write("\tParent Process: {0} PPID: {1}\n".format(parent_name, ppid))
            outfd.write("\tCreation Time: {0}\n".format(create_time))
            outfd.write("\tProcess Base Name(PEB): {0}\n".format(mod_basename))
            outfd.write("\tCommand Line(PEB): {0}\n".format(proc_cmd_line))
            outfd.write("\tHollow Type: {0}\n".format(hollow_types[hol_type]))
            outfd.write("\n")
            outfd.write( "VAD and PEB Comparison:\n")
            outfd.write( "\tBase Address(VAD): {0:#x}\n".format(vad_baseaddr))
            outfd.write( "\tProcess Path(VAD): {0}\n".format(vad_filename))
            outfd.write( "\tVad Protection: {0}\n".format(vad_protection))
            outfd.write( "\tVad Tag: {0}\n".format(vad_tag))
            outfd.write("\n")
            
            if hol_type == 0:
                addr_space = proc.get_process_address_space()
                dos_header = obj.Object("_IMAGE_DOS_HEADER", offset=proc_image_baseaddr, vm=addr_space)
                nt_header = dos_header.get_nt_header()
                optional_header = obj.Object("_IMAGE_OPTIONAL_HEADER", offset=nt_header.obj_offset+0x18, vm=addr_space)
                ep_addr = proc_image_baseaddr + optional_header.AddressOfEntryPoint
                content = addr_space.read(ep_addr, 64)
                outfd.write("\tBase Address(PEB): {0:#x}\n".format(proc_image_baseaddr))
                outfd.write("\tProcess Path(PEB): {0}\n" .format(mod_fullname))
                outfd.write("\tMemory Protection: {0}\n".format(vad_protection))
                outfd.write("\tMemory Tag: {0}\n".format(vad_tag))
                outfd.write("\n")
                outfd.write("Disassembly(Entry Point):\n")
                if content != None:
                    outfd.write("\n".join(["\t{0:#010x} {1:<16} {2}".format(o, h, i) 
                     for o, i, h in malfind.Disassemble(content, ep_addr)
                    ]))
                else:
                    outfd.write("\tNo Disassembly: Memory Unreadable at {0:#010x}\n".format(ep_addr))
                    
                outfd.write("\n\n")
            
            if (hol_type == 1) or (hol_type == 2):
                for vad, addr_space in proc.get_vads():
                    if vad.Start == proc_image_baseaddr:
                        content = addr_space.read(vad.Start, 64)
                        outfd.write("\tBase Address(PEB): {0:#x}\n".format(proc_image_baseaddr))
                        outfd.write("\tProcess Path(PEB): {0}\n" .format(mod_fullname))
                        outfd.write("\tMemory Protection: {0}\n".format(str(vadinfo.PROTECT_FLAGS.get(vad.VadFlags.Protection.v()) or  "")))
                        outfd.write("\tMemory Tag: {0}\n".format(str(vad.Tag or "")))
                        outfd.write("\n")
                        if content != None:
                            outfd.write("".join(["{0:#010x}  {1:<48}  {2}\n".format(vad.Start + o, h, ''.join(c))
                                       for o, h, c in utils.Hexdump(content)]))
                        else:
                            outfd.write("\tNo Hexdump: Memory Unreadable at {0:#010x}\n".format(vad.Start))
                        outfd.write("\n")
                        
            outfd.write("Similar Processes:\n")
            for similar_proc in similar_procs:
                (process_name, process_id, parent_name, parent_id, creation_time) = similar_proc
                outfd.write("\t{0}({1}) Parent:{2}({3}) Start:{4}\n".format(process_name,
                                                                            process_id,
                                                                            parent_name,
                                                                            parent_id,
                                                                            creation_time))
            outfd.write("\n")
            
            outfd.write("Suspicious Memory Regions:\n")
            for vad, addr_space in proc.get_vads():
                content = addr_space.read(vad.Start, 64)
                if content == None:
                    continue
                vad_prot = str(vadinfo.PROTECT_FLAGS.get(vad.VadFlags.Protection.v()))
                if obj.Object("_IMAGE_DOS_HEADER", offset = vad.Start, vm = addr_space).e_magic != 0x5A4D:
                    flag = "No PE/Possibly Code"
                    if (vad_prot == "PAGE_EXECUTE_READWRITE"):
                        sus_addr = vad.Start
                        outfd.write("\t{0:#x}({1})  Protection: {2}  Tag: {3}\n".format(vad.Start,
                                                                                        flag,
                                                                                        vad_prot,
                                                                                        str(vad.Tag or "")))
                        if self._config.DUMP_DIR:
                            filename = os.path.join(self._config.DUMP_DIR,"process.{0}.{1:#x}.dmp".format(hol_pid, sus_addr))
                            self.dump_vad(filename, vad, addr_space)
                    
                    elif (vad_prot == "PAGE_EXECUTE_WRITECOPY"):
                        sus_addr = vad.Start
                        outfd.write("\t{0:#x}({1})  Protection: {2}  Tag: {3}\n".format(sus_addr,
                                                                                        flag,
                                                                                        vad_prot,
                                                                                        str(vad.Tag or "")))
                        if self._config.DUMP_DIR:
                            filename = os.path.join(self._config.DUMP_DIR,"process.{0}.{1:#x}.dmp".format(hol_pid, 
                                                                                                          sus_addr))
                            self.dump_vad(filename, vad, addr_space)
                        
                else:
                    if vad_prot == "PAGE_EXECUTE_READWRITE":
                        flag = "PE Found"
                        sus_addr = vad.Start
                        outfd.write("\t{0:#x}({1})  Protection: {2}  Tag: {3}\n".format(sus_addr,
                                                                                        flag,
                                                                                        vad_prot,
                                                                                        str(vad.Tag or "")))
                        if self._config.DUMP_DIR:
                            filename = os.path.join(self._config.DUMP_DIR,"process.{0}.{1:#x}.dmp".format(hol_pid, 
                                                                                                          sus_addr))
                            self.dump_vad(filename, vad, addr_space)
                    
                    elif (vad_prot == "PAGE_EXECUTE_WRITECOPY") and (not bool(vad.FileObject)):
                        flag = "PE - No Mapped File"
                        sus_addr = vad.Start
                        outfd.write("\t{0:#x}({1})  Protection: {2}  Tag: {3}\n".format(sus_addr,
                                                                                        flag,
                                                                                        vad_prot,
                                                                                        str(vad.Tag or "")))
                        if self._config.DUMP_DIR:
                            filename = os.path.join(self._config.DUMP_DIR,"process.{0}.{1:#x}.dmp".format(hol_pid, 
                                                                                                          sus_addr))
                            self.dump_vad(filename, vad, addr_space)
                        
            outfd.write("---------------------------------------------------\n\n")