Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
408 lines (353 sloc) 20.1 KB
import struct
import volatility
import inspect
import importlib
import volatility.plugins.common as common
from volatility.renderers import TreeGrid
from volatility.renderers.basic import Address
from volatility import utils, obj, win32
from volatility.plugins.malware import malfind
from volatility.plugins.malware.timers import Timers
from unicorn import *
from unicorn.x86_const import *
class timerResult():
def __init__(self, process, thread, timerRoutine):
self.thread = thread
self.process = process
self.routine = timerRoutine
self.didROP = "Unknown"
self.didAdjustPerms = "Unknown"
self.didJumpToAdjusted = "Unknown"
self.adjustedAddresses = []
self.probablePayload = 0
self.prolog = "Unknown"
instrStream = process.get_process_address_space().read(timerRoutine, 16)
if not instrStream:
print "Process %s '%s': Can't read instruction stream at %s; perhaps it is paged out" % (hex(int(process.obj_offset)), process.ImageFileName, hex(timerRoutine))
self.prolog = disAssembleInstr(instrStream, timerRoutine, 5)
def disAssembleInstr(instrStream, instrLocation, opcodeCount):
toRet = []
for _, i, _ in malfind.Disassemble(instrStream, instrLocation, True):
opcodeCount = opcodeCount - 1
if opcodeCount == 0:
return "; ".join(toRet)
# Unicorn doesn't support virtual memory properly (see unicorn bug #947) so there's not really much point setting up
# GDT/etc.
def printUnicornContext(pas, nextIns, unicornEng):
print "Unicorn context: "
instrStream =, 20)
for _, i, _ in malfind.Disassemble(instrStream, nextIns):
print "\t %s" % i
print "ESP = %s" % hex(unicornEng.reg_read(UC_X86_REG_ESP))
print "EAX = %s" % hex(unicornEng.reg_read(UC_X86_REG_EAX))
print "EBX = %s" % hex(unicornEng.reg_read(UC_X86_REG_EBX))
print "ECX = %s" % hex(unicornEng.reg_read(UC_X86_REG_ECX))
print "EDX = %s" % hex(unicornEng.reg_read(UC_X86_REG_EDX))
print "ESI = %s" % hex(unicornEng.reg_read(UC_X86_REG_ESI))
print "EDI = %s" % hex(unicornEng.reg_read(UC_X86_REG_EDI))
print "EFLAGS = %s" % hex(unicornEng.reg_read(UC_X86_REG_EFLAGS))
print ""
def getWoW64Profile(process):
Given a 64-bit Windows process, find the 32-bit profile which most closely matches the 64-bit proceesses profile.
Hopefully, this will be close enough enable us to decode 32-bit objects running under WoW.
pas = process.get_process_address_space()
wow64ProfileFull = str(pas.profile.__class__).strip('><\'')[len('class: '):]
wow64ProfileNameSpace = '.'.join(wow64ProfileFull.split('.')[:-1])
wow64ProfileName = str(wow64ProfileFull.split('.')[-1:][0].replace('x64', 'x86'))
module = importlib.import_module(wow64ProfileNameSpace)
class_ = getattr(module, wow64ProfileName)
return class_()
def isKernelSpace(process, DllBase):
if process.get_process_address_space().profile.metadata.get("memory_model") == "32bit":
# TODO: support 3GB address mode, if it's worth it
return DllBase < 0x80000000
return DllBase < 0x8000000000000000
# It seems that certain SPs of certain OSs (vista and certain versions of 2003/XP64) use a non-standard
# APC format when running in WoW64 mode. We will apply this type only if we detect a version of windows
# which uses it.
WoW64UnusualAPC = {
'_KAPC_WOW64' : [ 0x58, {
'Type' : [ 0x0, ['unsigned char']],
'SpareByte0' : [ 0x1, ['unsigned char']],
'Size' : [ 0x2, ['unsigned char']],
'SpareByte1' : [ 0x3, ['unsigned char']],
'SpareLong0' : [ 0x4, ['unsigned long']],
'Thread' : [ 0x8, ['pointer64', ['_KTHREAD']]],
'ApcListEntry' : [ 0x10, ['_LIST_ENTRY']],
'KernelRoutine' : [ 0x20, ['pointer64', ['void']]],
'RundownRoutine' : [ 0x28, ['pointer64', ['void']]],
'unknown' : [ 0x30, ['pointer64', ['void']]],
'NormalContext' : [ 0x38, ['pointer32', ['void']]],
'NormalRoutine' : [ 0x3C, ['pointer32', ['void']]],
'SystemArgument1' : [ 0x40, ['pointer64', ['void']]],
'SystemArgument2' : [ 0x48, ['pointer64', ['void']]],
'ApcStateIndex' : [ 0x50, ['unsigned char']],
'ApcMode' : [ 0x51, ['unsigned char']],
'Inserted' : [ 0x52, ['unsigned char']],
class APCVTypes(obj.ProfileModification):
before = ['WindowsOverlay']
conditions = {'os': lambda x: x == 'windows'}
def modification(self, profile):
if profile.metadata.get("memory_model", "32bit") == "64bit":
class gargoyle(common.AbstractWindowsCommand):
def __init__(self, config, *args, **kwargs):
common.AbstractWindowsCommand.__init__(self, config, *args, **kwargs)
config.add_option("VERBOSE", default=0, action='count',
short_option='v', help='Verbose information')
config.add_option("ALLTIMERS", default = False, action = 'store_true',
help = "Do not filter timers by valid EPROCESS")
def dbgMsg(self, *args):
if self._config.VERBOSE > 0:
print " ".join(map(str, args))
# This is called when Unicorn needs to access some memory that isn't mapped yet.
# We simply map the memory, copy in its contents from the debuggee, and return.
# Our main loop will retry. We signal errors by setting self.emulationFaulted.
def badmemWrapped(self, uc, access, address, size, value, user_data):
self.dbgMsg("Access to unmapped memory %s" % hex(address))
if self.pas == None:
self.dbgMsg("Unable to handle memory mapping with no active process")
raise MemoryError
# Unicorn will only successfully map page-aligned addresses, so map the whole page.
pageSize = 0x1000
pageBase = address & (~(pageSize-1))
uc.mem_map(pageBase, pageSize)
# Read from the debuggee..
pageCts =, pageSize)
if pageCts == None:
self.dbgMsg ("Unable to read %s bytes at %s" % (hex(pageSize), hex(pageBase)))
raise MemoryError
# And write to Unicorn.
uc.mem_write(pageBase, pageCts)
self.dbgMsg( "Mapped %s bytes at base %s" % (hex(pageSize), hex(pageBase)))
return True
def badmem(self, uc, access, address, size, value, user_data):
return self.badmemWrapped(uc, access, address, size, value, user_data)
except Exception as e:
self.emulationFaulted = e
def unified_output(self, data):
return TreeGrid([("Process", str),
("Handler", Address),
("Prolog", str),
("Adjusted page permissions", str),
("Branched to code after altering page permission", str),
("Probable payload", Address)],
def generator(self, data):
for timer in data:
yield(0, [ str(timer.process.ImageFileName),
def calculate(self):
self.exportCache = {}
addr_space = utils.load_as(self._config)
if self._config.ALLTIMERS:
validProcesses = None
validProcesses = []
for p in win32.tasks.pslist(addr_space):
# Firstly, we must iterate over the timers found by the timers plugin. We can then find the owning process, the
# associated APC, and the thread. We need the process because we will need to know what process context the APC's
# NormalRoutine (if any) will run in, in order to do our emulation.
for timeraddress, _ in Timers(self._config).calculate():
timer = obj.Object("_ETIMER", offset=timeraddress.obj_offset, vm=timeraddress.obj_vm)
self.dbgMsg("Timer at %s dpc %s" % (hex(int(timer.obj_offset)), hex(int(timeraddress.Dpc.DeferredRoutine))))
etimer = timer.cast("_ETIMER")
# Check for the weird APC format
version = (addr_space.profile.metadata.get('major', 0),
addr_space.profile.metadata.get('minor', 0))
if addr_space.profile.metadata.get("memory_model") == "64bit" and ((version == (5, 2) ) or (version == (6, 0))):
apc = obj.Object("_KAPC_WOW64", offset=etimer.TimerApc.obj_offset, vm=etimer.TimerApc.obj_vm)
apc = obj.Object("_KAPC" , offset=etimer.TimerApc.obj_offset, vm=etimer.TimerApc.obj_vm)
routine = int(apc.NormalRoutine)
thread = obj.Object("_ETHREAD", offset=int(apc.Thread), vm=addr_space)
if (thread == None) | (routine == 0):
# This APC has no user-mode payload.
process = thread.owning_process()
if process == None:
# This usually happens when a timer is not pointing to a valid thread. I'm not sure why this happens -
# I guess there's some flag in the timer which states that it isn't valid, or the timer/timer list is
# # being manipulated when we dump.
self.dbgMsg('Timer %s : warning: Thread ID %s has no owning process, skipping' % (hex(int(timeraddress.obj_offset)), hex(int(thread.Cid.UniqueThread))))
# If this is a WoW64 APC - ie, an APC queued by a 32-bit thread on a 64-bit windows install - then we must
# 'decode' the NormalRoutine by shifting and negating it.
# We detect these WoW64-style APCs by comparing the top half of the 64bit address, except bit zero, to 0xffffffff. I'm not
# sure if this is reliable, but it seems to work.
if (((routine >> 32) | 0x01) == 0xffffffff):
routine32bit = (-(routine >> 2)) & 0xffffffff
self.dbgMsg("WoW64-style APC routine decoded %s to %s" % (hex(routine), hex(routine32bit)))
routine = routine32bit
if (validProcesses != None) and (not process in validProcesses):
for result in self.examine(addr_space, process, thread, routine, apc, timer):
yield result
def getWoW64Modules(self, process):
pas = process.get_process_address_space()
# Get a pointer to the 32-bit PEB, which is stored as part of the _EWOW64PROCESS. This is of type nt!_PEB32.
peb32 = process.Peb32
ldr = int(peb32.Ldr)
# Since this PEB is 32bit, we must use definitions from the 32bit version of windows which corresponds to this
# build.
profile32 = getWoW64Profile(process)
pas32 = pas
pas32.profile = profile32
pebLdrData = obj.Object("_PEB_LDR_DATA", offset=ldr, vm=pas32)
modules = pebLdrData.InMemoryOrderModuleList.cast("_LIST_ENTRY").list_of_type("_LDR_DATA_TABLE_ENTRY", "InMemoryOrderLinks")
return modules
def findExport(self, process, addr_space, moduleName, exportName):
# mind the case! 'WoW' vs 'Wow'.
if hasattr(process.m, 'Wow64Process'):
wow64Process = process.Wow64Process
elif hasattr(process, 'WoW64Process'):
wow64Process = process.WoW64Process
wow64Process = None
isWoW64 = False
if wow64Process != None and wow64Process.v() != 0:
isWoW64 = True
if isWoW64:
# WoW64 processes are treated specially, since we must get 32bit modules via the 32bit PEB.
pas = process.get_process_address_space()
if pas == None:
# Probably not a real EPROCESS.
return None
modList = self.getWoW64Modules(process)
# Not a WoW64 process, so just get the modules normally.
modList = process.get_mem_modules()
exp = self.findExportInModuleList(modList, moduleName, exportName, process, isWoW64)
if exp == None:
print "Unable to find export %s!%s in process %s" % (moduleName, exportName, process.ImageFileName)
return exp
def findExportInModuleList(self, moduleList, moduleName, exportName, process, isWoW64):
moduleNameLowercase = moduleName.lower()
exportNameLowercase = exportName.lower()
for m in moduleList:
dllName = str(m.BaseDllName).lower()
if dllName == moduleNameLowercase:
# Cache per-process (since a module may appear in a different process at a different base). For kernel modules
# there's no need to be per-process, so don't bother.
if isKernelSpace(process, m.DllBase):
cacheKey = "%s!%s (kernel)" % (moduleNameLowercase, exportNameLowercase)
cacheKey = "%s!%s %s (wow64 %s)" % (moduleNameLowercase, exportNameLowercase, hex(m.DllBase), isWoW64)
if cacheKey in self.exportCache.keys():
return self.exportCache[cacheKey]
for _, expAddress, expName in m.exports():
if str(expName).lower() == exportNameLowercase:
toReturn = m.DllBase + expAddress
self.dbgMsg("Found %s ! %s at %s (%s)" % (m.BaseDllName, expName, hex(toReturn), hex(m.DllBase)))
self.exportCache[cacheKey] = toReturn
return toReturn
return None
def examine(self, addr_space, process, thread, routine, apc, timer):
# We will now emulate through the instruction stream, starting at the APC handler, and see if anything fishy
# goes on. Specifically, we will see if the APC calls VirtualProtect. If it does, we will see if it also
# tries to jump to the newly-VirtualProtect'ed memory - a sure sign of Gargoyle-ness.
VirtualProtect = self.findExport(process, addr_space, "KERNEL32.DLL", "VirtualProtect")
VirtualProtectEx = self.findExport(process, addr_space, "KERNEL32.DLL", "VirtualProtectEx")
# We'll need to set the process address space so that our badmem callback can use it later on.
self.pas = process.get_process_address_space()
self.emulationFaulted = None
result = timerResult(process, thread, routine)
self.dbgMsg("Timer %s APC %s routine %s in process %s ('%s') thread %s" % (hex(int(timer.obj_offset)), hex(int(apc.obj_offset)), hex(routine), hex(int(process.obj_offset)), process.ImageFileName, hex(thread.StartAddress)))
unicornEng = Uc(UC_ARCH_X86, UC_MODE_32)
# Populate the context from which to start emulating.
# We use an arbitrary ESP, with a magic value to signify that the APC handler has returned.
initialStackBase = 0x00000000f0000000
unicornEng.mem_map(initialStackBase, 2 * 1024 * 1024)
unicornEng.mem_write(initialStackBase + 0x100 + 0, "\xbe\xba\xde\xc0")
# We push the argument which the APC handler is given
unicornEng.mem_write(initialStackBase + 0x100 + 4,, 4))
unicornEng.reg_write(UC_X86_REG_ESP, initialStackBase + 0x100)
# Set up our handlers, which will map memory on-demand from the debuggee
unicornEng.hook_add(UC_HOOK_MEM_READ_UNMAPPED, self.badmem)
unicornEng.hook_add(UC_HOOK_MEM_WRITE_INVALID, self.badmem)
unicornEng.hook_add(UC_HOOK_MEM_FETCH_UNMAPPED, self.badmem)
# There's not really much point mapping the GDT, since Unicorn doesn't properly support paging.
# See Unicorn's bug #947, "(x86) Emulated CPU is not translating virtual memory addresses".
# Now, lets emulate some instructions! We won't get many, because Unicorn can't emulate a lot of things (like
# segment-prefixed instructions, as used by wow64) but we'll get enough to detect most ROP chains.
instrEmulated = 0
nextIns = routine
memoryRange = None
while instrEmulated < 10000:
if self._config.VERBOSE > 0:
print "Before instruction %d at %s:" % (instrEmulated, hex(nextIns))
print "CS:IP = %s:%s SS:SP = %s:%s" % (
hex(unicornEng.reg_read(UC_X86_REG_CS)), hex(unicornEng.reg_read(UC_X86_REG_EIP)),
hex(unicornEng.reg_read(UC_X86_REG_SS)), hex(unicornEng.reg_read(UC_X86_REG_ESP)))
instrStream =, 15)
for _, i, _ in malfind.Disassemble(instrStream, nextIns):
print "\t%s\t%s" % (hex(nextIns), i)
# Attempt to emulate a single instruction
unicornEng.emu_start(nextIns, nextIns + 0x10, count = 1)
except unicorn.UcError as e1:
if self.emulationFaulted != None:
# Great, we emulated an instruction. Move on to the next instruction.
nextIns = unicornEng.reg_read(UC_X86_REG_EIP)
instrEmulated = instrEmulated + 1
# If we're now at our magic address, then our APC has completed executing entirely. That's all, folks.
if nextIns == 0xc0debabe:
# Now we can check for some suspicious circumstance.
esp = unicornEng.reg_read(UC_X86_REG_ESP)
if esp == int(apc.NormalContext):
result.didROP = "True"
self.dbgMsg("APC has performed stack pivot; new stack is its context pointer")
if VirtualProtect == None:
# If we didn't find VirtualProtect, we can't go any further. I guess a stack pivot is a pretty big
# red flag anyway.
if VirtualProtectEx != None:
if nextIns == VirtualProtectEx:
result.didAdjustPerms = "True"
# Read the arguments to VirtualProtect, and the return address, from the stack
returnAddress = struct.unpack("I", unicornEng.mem_read(esp - 0, 4))[0]
memoryRange = struct.unpack("I", unicornEng.mem_read(esp + 8, 4))[0]
self.dbgMsg("VirtualProtectEx: Timer routine is adjusting memory permissions of range %s" % hex(memoryRange))
# Set the return address to whatever VirtualProtect would've returned to
nextIns = returnAddress
unicornEng.reg_write(UC_X86_REG_EIP, returnAddress)
# Pop five args plus the return address off the (32bit) stack
unicornEng.reg_write(UC_X86_REG_ESP, esp + (6*4))
if VirtualProtect != None:
if nextIns == VirtualProtect:
result.didAdjustPerms = "True"
# Read the arguments to VirtualProtect, and the return address, from the stack
returnAddress = struct.unpack("I", unicornEng.mem_read(esp - 0, 4))[0]
memoryRange = struct.unpack("I", unicornEng.mem_read(esp + 4, 4))[0]
self.dbgMsg("VirtualProtect: Timer routine is adjusting memory permissions of range %s" % hex(memoryRange))
# Set the return address to whatever VirtualProtect would've returned to
nextIns = returnAddress
unicornEng.reg_write(UC_X86_REG_EIP, returnAddress)
# Pop four args plus the return address off the (32bit) stack
unicornEng.reg_write(UC_X86_REG_ESP, esp + (5 * 4))
if nextIns in result.adjustedAddresses:
result.didJumpToAdjusted = "True"
result.probablePayload = nextIns
self.dbgMsg( "Timer routine is jumping to newly-executable code at %s!" % hex(memoryRange))
yield result