import Unicorn2

This commit is contained in:
Nguyen Anh Quynh
2021-10-03 22:14:44 +08:00
parent 772558119a
commit aaaea14214
837 changed files with 368717 additions and 200912 deletions

View File

@ -3,12 +3,12 @@
import ctypes
import ctypes.util
import distutils.sysconfig
from functools import wraps
import pkg_resources
import inspect
import os.path
import sys
import weakref
import functools
from . import x86_const, arm64_const, unicorn_const as uc
@ -105,6 +105,8 @@ def _setup_prototype(lib, fname, restype, *argtypes):
getattr(lib, fname).argtypes = argtypes
ucerr = ctypes.c_int
uc_mode = ctypes.c_int
uc_arch = ctypes.c_int
uc_engine = ctypes.c_void_p
uc_context = ctypes.c_void_p
uc_hook_h = ctypes.c_size_t
@ -130,6 +132,7 @@ _setup_prototype(_uc, "uc_mem_write", ucerr, uc_engine, ctypes.c_uint64, ctypes.
_setup_prototype(_uc, "uc_emu_start", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_size_t)
_setup_prototype(_uc, "uc_emu_stop", ucerr, uc_engine)
_setup_prototype(_uc, "uc_hook_del", ucerr, uc_engine, uc_hook_h)
_setup_prototype(_uc, "uc_mmio_map", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_size_t, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p)
_setup_prototype(_uc, "uc_mem_map", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_size_t, ctypes.c_uint32)
_setup_prototype(_uc, "uc_mem_map_ptr", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_size_t, ctypes.c_uint32, ctypes.c_void_p)
_setup_prototype(_uc, "uc_mem_unmap", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_size_t)
@ -140,12 +143,12 @@ _setup_prototype(_uc, "uc_free", ucerr, ctypes.c_void_p)
_setup_prototype(_uc, "uc_context_save", ucerr, uc_engine, uc_context)
_setup_prototype(_uc, "uc_context_restore", ucerr, uc_engine, uc_context)
_setup_prototype(_uc, "uc_context_size", ctypes.c_size_t, uc_engine)
_setup_prototype(_uc, "uc_context_reg_read", ucerr, uc_context, ctypes.c_int, ctypes.c_void_p)
_setup_prototype(_uc, "uc_context_reg_write", ucerr, uc_context, ctypes.c_int, ctypes.c_void_p)
_setup_prototype(_uc, "uc_context_free", ucerr, uc_context)
_setup_prototype(_uc, "uc_mem_regions", ucerr, uc_engine, ctypes.POINTER(ctypes.POINTER(_uc_mem_region)), ctypes.POINTER(ctypes.c_uint32))
# uc_hook_add is special due to variable number of arguments
_uc.uc_hook_add = _uc.uc_hook_add
_uc.uc_hook_add.restype = ucerr
# https://bugs.python.org/issue42880
_setup_prototype(_uc, "uc_hook_add", ucerr, uc_engine, ctypes.POINTER(uc_hook_h), ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint64, ctypes.c_uint64)
UC_HOOK_CODE_CB = ctypes.CFUNCTYPE(None, uc_engine, ctypes.c_uint64, ctypes.c_size_t, ctypes.c_void_p)
UC_HOOK_INSN_INVALID_CB = ctypes.CFUNCTYPE(ctypes.c_bool, uc_engine, ctypes.c_void_p)
@ -168,7 +171,12 @@ UC_HOOK_INSN_OUT_CB = ctypes.CFUNCTYPE(
ctypes.c_int, ctypes.c_uint32, ctypes.c_void_p
)
UC_HOOK_INSN_SYSCALL_CB = ctypes.CFUNCTYPE(None, uc_engine, ctypes.c_void_p)
UC_MMIO_READ_CB = ctypes.CFUNCTYPE(
ctypes.c_uint64, uc_engine, ctypes.c_uint64, ctypes.c_int, ctypes.c_void_p
)
UC_MMIO_WRITE_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.c_uint64, ctypes.c_int, ctypes.c_uint64, ctypes.c_void_p
)
# access to error code via @errno of UcError
class UcError(Exception):
@ -199,27 +207,103 @@ def version_bind():
def uc_arch_supported(query):
return _uc.uc_arch_supported(query)
# uc_reg_read/write and uc_context_reg_read/write.
def reg_read(reg_read_func, arch, reg_id, opt=None):
if arch == uc.UC_ARCH_X86:
if reg_id in [x86_const.UC_X86_REG_IDTR, x86_const.UC_X86_REG_GDTR, x86_const.UC_X86_REG_LDTR, x86_const.UC_X86_REG_TR]:
reg = uc_x86_mmr()
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.selector, reg.base, reg.limit, reg.flags
if reg_id in range(x86_const.UC_X86_REG_FP0, x86_const.UC_X86_REG_FP0+8):
reg = uc_x86_float80()
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.mantissa, reg.exponent
if reg_id in range(x86_const.UC_X86_REG_XMM0, x86_const.UC_X86_REG_XMM0+8):
reg = uc_x86_xmm()
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.low_qword | (reg.high_qword << 64)
if reg_id in range(x86_const.UC_X86_REG_YMM0, x86_const.UC_X86_REG_YMM0+16):
reg = uc_x86_ymm()
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.first_qword | (reg.second_qword << 64) | (reg.third_qword << 128) | (reg.fourth_qword << 192)
if reg_id is x86_const.UC_X86_REG_MSR:
if opt is None:
raise UcError(uc.UC_ERR_ARG)
reg = uc_x86_msr()
reg.rid = opt
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.value
def _catch_hook_exception(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
"""Catches exceptions raised in hook functions.
if arch == uc.UC_ARCH_ARM64:
if reg_id in range(arm64_const.UC_ARM64_REG_Q0, arm64_const.UC_ARM64_REG_Q31+1) or range(arm64_const.UC_ARM64_REG_V0, arm64_const.UC_ARM64_REG_V31+1):
reg = uc_arm64_neon128()
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.low_qword | (reg.high_qword << 64)
If an exception is raised, it is saved to the Uc object and a call to stop
emulation is issued.
"""
try:
return func(self, *args, **kwargs)
except Exception as e:
# If multiple hooks raise exceptions, just use the first one
if self._hook_exception is None:
self._hook_exception = e
# read to 64bit number to be safe
reg = ctypes.c_uint64(0)
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.value
self.emu_stop()
def reg_write(reg_write_func, arch, reg_id, value):
reg = None
return wrapper
if arch == uc.UC_ARCH_X86:
if reg_id in [x86_const.UC_X86_REG_IDTR, x86_const.UC_X86_REG_GDTR, x86_const.UC_X86_REG_LDTR, x86_const.UC_X86_REG_TR]:
assert isinstance(value, tuple) and len(value) == 4
reg = uc_x86_mmr()
reg.selector = value[0]
reg.base = value[1]
reg.limit = value[2]
reg.flags = value[3]
if reg_id in range(x86_const.UC_X86_REG_FP0, x86_const.UC_X86_REG_FP0+8):
reg = uc_x86_float80()
reg.mantissa = value[0]
reg.exponent = value[1]
if reg_id in range(x86_const.UC_X86_REG_XMM0, x86_const.UC_X86_REG_XMM0+8):
reg = uc_x86_xmm()
reg.low_qword = value & 0xffffffffffffffff
reg.high_qword = value >> 64
if reg_id in range(x86_const.UC_X86_REG_YMM0, x86_const.UC_X86_REG_YMM0+16):
reg = uc_x86_ymm()
reg.first_qword = value & 0xffffffffffffffff
reg.second_qword = (value >> 64) & 0xffffffffffffffff
reg.third_qword = (value >> 128) & 0xffffffffffffffff
reg.fourth_qword = value >> 192
if reg_id is x86_const.UC_X86_REG_MSR:
reg = uc_x86_msr()
reg.rid = value[0]
reg.value = value[1]
if arch == uc.UC_ARCH_ARM64:
if reg_id in range(arm64_const.UC_ARM64_REG_Q0, arm64_const.UC_ARM64_REG_Q31+1) or range(arm64_const.UC_ARM64_REG_V0, arm64_const.UC_ARM64_REG_V31+1):
reg = uc_arm64_neon128()
reg.low_qword = value & 0xffffffffffffffff
reg.high_qword = value >> 64
if reg is None:
# convert to 64bit number to be safe
reg = ctypes.c_uint64(value)
status = reg_write_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return
class uc_x86_mmr(ctypes.Structure):
"""Memory-Management Register for instructions IDTR, GDTR, LDTR, TR."""
@ -306,6 +390,8 @@ class Uc(object):
def __init__(self, arch, mode):
# verify version compatibility with the core before doing anything
(major, minor, _combined) = uc_version()
# print("core version =", uc_version())
# print("binding version =", uc.UC_API_MAJOR, uc.UC_API_MINOR)
if major != uc.UC_API_MAJOR or minor != uc.UC_API_MINOR:
self._uch = None
# our binding version is different from the core's API version
@ -319,10 +405,9 @@ class Uc(object):
raise UcError(status)
# internal mapping table to save callback & userdata
self._callbacks = {}
self._ctype_cbs = {}
self._ctype_cbs = []
self._callback_count = 0
self._cleanup.register(self)
self._hook_exception = None # The exception raised in a hook
@staticmethod
def release_handle(uch):
@ -340,9 +425,6 @@ class Uc(object):
if status != uc.UC_ERR_OK:
raise UcError(status)
if self._hook_exception is not None:
raise self._hook_exception
# stop emulation
def emu_stop(self):
status = _uc.uc_emu_stop(self._uch)
@ -351,100 +433,11 @@ class Uc(object):
# return the value of a register
def reg_read(self, reg_id, opt=None):
if self._arch == uc.UC_ARCH_X86:
if reg_id in [x86_const.UC_X86_REG_IDTR, x86_const.UC_X86_REG_GDTR, x86_const.UC_X86_REG_LDTR, x86_const.UC_X86_REG_TR]:
reg = uc_x86_mmr()
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.selector, reg.base, reg.limit, reg.flags
if reg_id in range(x86_const.UC_X86_REG_FP0, x86_const.UC_X86_REG_FP0+8):
reg = uc_x86_float80()
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.mantissa, reg.exponent
if reg_id in range(x86_const.UC_X86_REG_XMM0, x86_const.UC_X86_REG_XMM0+8):
reg = uc_x86_xmm()
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.low_qword | (reg.high_qword << 64)
if reg_id in range(x86_const.UC_X86_REG_YMM0, x86_const.UC_X86_REG_YMM0+16):
reg = uc_x86_ymm()
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.first_qword | (reg.second_qword << 64) | (reg.third_qword << 128) | (reg.fourth_qword << 192)
if reg_id is x86_const.UC_X86_REG_MSR:
if opt is None:
raise UcError(uc.UC_ERR_ARG)
reg = uc_x86_msr()
reg.rid = opt
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.value
if self._arch == uc.UC_ARCH_ARM64:
if reg_id in range(arm64_const.UC_ARM64_REG_Q0, arm64_const.UC_ARM64_REG_Q31+1) or range(arm64_const.UC_ARM64_REG_V0, arm64_const.UC_ARM64_REG_V31+1):
reg = uc_arm64_neon128()
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.low_qword | (reg.high_qword << 64)
# read to 64bit number to be safe
reg = ctypes.c_uint64(0)
status = _uc.uc_reg_read(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.value
return reg_read(functools.partial(_uc.uc_reg_read, self._uch), self._arch, reg_id, opt)
# write to a register
def reg_write(self, reg_id, value):
reg = None
if self._arch == uc.UC_ARCH_X86:
if reg_id in [x86_const.UC_X86_REG_IDTR, x86_const.UC_X86_REG_GDTR, x86_const.UC_X86_REG_LDTR, x86_const.UC_X86_REG_TR]:
assert isinstance(value, tuple) and len(value) == 4
reg = uc_x86_mmr()
reg.selector = value[0]
reg.base = value[1]
reg.limit = value[2]
reg.flags = value[3]
if reg_id in range(x86_const.UC_X86_REG_FP0, x86_const.UC_X86_REG_FP0+8):
reg = uc_x86_float80()
reg.mantissa = value[0]
reg.exponent = value[1]
if reg_id in range(x86_const.UC_X86_REG_XMM0, x86_const.UC_X86_REG_XMM0+8):
reg = uc_x86_xmm()
reg.low_qword = value & 0xffffffffffffffff
reg.high_qword = value >> 64
if reg_id in range(x86_const.UC_X86_REG_YMM0, x86_const.UC_X86_REG_YMM0+16):
reg = uc_x86_ymm()
reg.first_qword = value & 0xffffffffffffffff
reg.second_qword = (value >> 64) & 0xffffffffffffffff
reg.third_qword = (value >> 128) & 0xffffffffffffffff
reg.fourth_qword = value >> 192
if reg_id is x86_const.UC_X86_REG_MSR:
reg = uc_x86_msr()
reg.rid = value[0]
reg.value = value[1]
if self._arch == uc.UC_ARCH_ARM64:
if reg_id in range(arm64_const.UC_ARM64_REG_Q0, arm64_const.UC_ARM64_REG_Q31+1) or range(arm64_const.UC_ARM64_REG_V0, arm64_const.UC_ARM64_REG_V31+1):
reg = uc_arm64_neon128()
reg.low_qword = value & 0xffffffffffffffff
reg.high_qword = value >> 64
if reg is None:
# convert to 64bit number to be safe
reg = ctypes.c_uint64(value)
status = _uc.uc_reg_write(self._uch, reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg_write(functools.partial(_uc.uc_reg_write, self._uch), self._arch, reg_id, value)
# read from MSR - X86 only
def msr_read(self, msr_id):
@ -468,6 +461,33 @@ class Uc(object):
if status != uc.UC_ERR_OK:
raise UcError(status)
def _mmio_map_read_cb(self, handle, offset, size, user_data):
(cb, data) = self._callbacks[user_data]
return cb(self, offset, size, data)
def _mmio_map_write_cb(self, handle, offset, size, value, user_data):
(cb, data) = self._callbacks[user_data]
cb(self, offset, size, value, data)
def mmio_map(self, address, size, read_cb, user_data_read, write_cb, user_data_write):
internal_read_cb = ctypes.cast(UC_MMIO_READ_CB(self._mmio_map_read_cb), UC_MMIO_READ_CB)
internal_write_cb = ctypes.cast(UC_MMIO_WRITE_CB(self._mmio_map_write_cb), UC_MMIO_WRITE_CB)
self._callback_count += 1
self._callbacks[self._callback_count] = (read_cb, user_data_read)
read_count = self._callback_count
self._callback_count += 1
self._callbacks[self._callback_count] = (write_cb, user_data_write)
write_count = self._callback_count
status = _uc.uc_mmio_map(self._uch, address, size, internal_read_cb, read_count, internal_write_cb, write_count)
if status != uc.UC_ERR_OK:
raise UcError(status)
# https://docs.python.org/3/library/ctypes.html#callback-functions
self._ctype_cbs.append(internal_read_cb)
self._ctype_cbs.append(internal_write_cb)
# map a range of memory
def mem_map(self, address, size, perms=uc.UC_PROT_ALL):
status = _uc.uc_mem_map(self._uch, address, size, perms)
@ -500,49 +520,41 @@ class Uc(object):
raise UcError(status)
return result.value
@_catch_hook_exception
def _hookcode_cb(self, handle, address, size, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
cb(self, address, size, data)
@_catch_hook_exception
def _hook_mem_invalid_cb(self, handle, access, address, size, value, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
return cb(self, access, address, size, value, data)
@_catch_hook_exception
def _hook_mem_access_cb(self, handle, access, address, size, value, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
cb(self, access, address, size, value, data)
@_catch_hook_exception
def _hook_intr_cb(self, handle, intno, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
cb(self, intno, data)
@_catch_hook_exception
def _hook_insn_invalid_cb(self, handle, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
return cb(self, data)
@_catch_hook_exception
def _hook_insn_in_cb(self, handle, port, size, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
return cb(self, port, size, data)
@_catch_hook_exception
def _hook_insn_out_cb(self, handle, port, size, value, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
cb(self, port, size, value, data)
@_catch_hook_exception
def _hook_insn_syscall_cb(self, handle, user_data):
# call user's callback with self object
(cb, data) = self._callbacks[user_data]
@ -615,7 +627,7 @@ class Uc(object):
)
# save the ctype function so gc will leave it alone.
self._ctype_cbs[self._callback_count] = cb
self._ctype_cbs.append(cb)
if status != uc.UC_ERR_OK:
raise UcError(status)
@ -631,7 +643,7 @@ class Uc(object):
h = 0
def context_save(self):
context = UcContext(self._uch)
context = UcContext(self._uch, self._arch, self._mode)
status = _uc.uc_context_save(self._uch, context.context)
if status != uc.UC_ERR_OK:
raise UcError(status)
@ -664,14 +676,16 @@ class Uc(object):
class UcContext:
def __init__(self, h):
def __init__(self, h, arch, mode):
self._context = uc_context()
self._size = _uc.uc_context_size(h)
self._to_free = True
status = _uc.uc_context_alloc(h, ctypes.byref(self._context))
if status != uc.UC_ERR_OK:
raise UcError(status)
self._arch = arch
self._mode = mode
@property
def context(self):
return self._context
@ -680,16 +694,34 @@ class UcContext:
def size(self):
return self._size
@property
def arch(self):
return self._arch
@property
def mode(self):
return self._mode
# return the value of a register
def reg_read(self, reg_id, opt=None):
return reg_read(functools.partial(_uc.uc_context_reg_read, self._context), self.arch, reg_id, opt)
# write to a register
def reg_write(self, reg_id, value):
return reg_write(functools.partial(_uc.uc_context_reg_write, self._context), self.arch, reg_id, value)
# Make UcContext picklable
def __getstate__(self):
return (bytes(self), self.size)
return (bytes(self), self.size, self.arch, self.mode)
def __setstate__(self, state):
self._size = state[1]
self._context = ctypes.cast(ctypes.create_string_buffer(state[0], self._size), uc_context)
# __init__ won'e be invoked, so we are safe to set it here.
self._to_free = False
self._arch = state[2]
self._mode = state[3]
def __bytes__(self):
return ctypes.string_at(self.context, self.size)
@ -708,6 +740,8 @@ def debug():
"sparc": uc.UC_ARCH_SPARC,
"m68k": uc.UC_ARCH_M68K,
"x86": uc.UC_ARCH_X86,
"riscv": uc.UC_ARCH_RISCV,
"ppc": uc.UC_ARCH_PPC,
}
all_archs = ""