mirror of
https://github.com/WithSecureLabs/volatility-plugins
synced 2026-06-21 13:46:34 +00:00
Countercept
408 lines
20 KiB
Python
408 lines
20 KiB
Python
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))
|
|
else:
|
|
self.prolog = disAssembleInstr(instrStream, timerRoutine, 5)
|
|
|
|
def disAssembleInstr(instrStream, instrLocation, opcodeCount):
|
|
toRet = []
|
|
for _, i, _ in malfind.Disassemble(instrStream, instrLocation, True):
|
|
toRet.append(i)
|
|
opcodeCount = opcodeCount - 1
|
|
if opcodeCount == 0:
|
|
break
|
|
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 = pas.read(nextIns, 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
|
|
else:
|
|
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":
|
|
profile.vtypes.update(WoW64UnusualAPC)
|
|
|
|
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 = self.pas.read(pageBase, 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):
|
|
try:
|
|
return self.badmemWrapped(uc, access, address, size, value, user_data)
|
|
except Exception as e:
|
|
self.emulationFaulted = e
|
|
raise
|
|
|
|
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)],
|
|
self.generator(data))
|
|
|
|
def generator(self, data):
|
|
for timer in data:
|
|
yield(0, [ str(timer.process.ImageFileName),
|
|
Address(timer.routine),
|
|
str(timer.prolog),
|
|
str(timer.didAdjustPerms),
|
|
str(timer.didJumpToAdjusted),
|
|
Address(timer.probablePayload)])
|
|
|
|
def calculate(self):
|
|
self.exportCache = {}
|
|
addr_space = utils.load_as(self._config)
|
|
|
|
if self._config.ALLTIMERS:
|
|
validProcesses = None
|
|
else:
|
|
validProcesses = []
|
|
for p in win32.tasks.pslist(addr_space):
|
|
validProcesses.append(int(p.obj_offset))
|
|
|
|
# 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)
|
|
else:
|
|
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.
|
|
continue
|
|
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))))
|
|
continue
|
|
|
|
# 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):
|
|
continue
|
|
|
|
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
|
|
else:
|
|
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)
|
|
else:
|
|
# 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)
|
|
else:
|
|
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, apc.NormalContext.obj_vm.read(apc.NormalContext.obj_offset, 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 = self.pas.read(nextIns, 15)
|
|
for _, i, _ in malfind.Disassemble(instrStream, nextIns):
|
|
print "\t%s\t%s" % (hex(nextIns), i)
|
|
break
|
|
|
|
# Attempt to emulate a single instruction
|
|
try:
|
|
unicornEng.emu_start(nextIns, nextIns + 0x10, count = 1)
|
|
except unicorn.UcError as e1:
|
|
break
|
|
if self.emulationFaulted != None:
|
|
break
|
|
|
|
# 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:
|
|
break
|
|
|
|
# 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.
|
|
break
|
|
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]
|
|
|
|
result.adjustedAddresses.append(memoryRange)
|
|
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]
|
|
|
|
result.adjustedAddresses.append(memoryRange)
|
|
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))
|
|
break
|
|
yield result
|