forked from monnappa22/HollowFind
/
hollowfind.py
351 lines (306 loc) · 18.6 KB
/
hollowfind.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
# Author: Monnappa K A
# Email : monnappa22@gmail.com
# Twitter: @monnappa22
# Description: Volatility plugin to detect different types of Process Hollowing
import os
import struct
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 _get_peb_imagebase(self, proc):
# for WOW64 processes, there are 2 PEBs: one from the 64 bit world, and one for 32.
# In regards to Process Hollowing, the 32 bit version must be considered.
# Since the 32 bit PEB is currently not loaded, we simply read the image base directly
if proc.WoW64Process:
peb_base = proc.WoW64Process.Peb
proc_addr_space = proc.get_process_address_space()
return struct.unpack('<I', proc_addr_space.read((peb_base + 8), 4))[0]
else:
return proc.Peb.ImageBaseAddress
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()
proc_cmd_line = proc.Peb.ProcessParameters.CommandLine
proc_image_baseaddr = -1
mod_baseaddr = -1
mod_size = -1
mod_basename = "N/A"
mod_fullname = "N/A"
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 = self._get_peb_imagebase(proc)
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 == self._get_peb_imagebase(proc)):
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")