Restore Py2 compatibility

This commit is contained in:
mio 2023-06-28 16:40:43 +08:00
parent d6d1af243f
commit 247d643f2f
No known key found for this signature in database
GPG Key ID: DFF27E34A47CB873
6 changed files with 2108 additions and 1112 deletions

View File

@ -1,41 +0,0 @@
This documentation explains how to install the python binding for Unicorn
from source.
1. Installing on Linux:
$ sudo python setup.py install
This will build the core C library, package it with the python bindings,
and install it to your system.
If you want to prevent the build of the native library during the python installation,
set the environment variable LIBUNICORN_PATH. You may also set this to a directory
containing libunicorn.so if you wish to use a verison of the native library other than
the globally installed one.
2. Installing on Windows:
Run the following command in command prompt:
C:\> C:\location_to_python\python.exe setup.py install
Next, copy all the DLL files from the 'Core engine for Windows' package available
on the Unicorn download page and paste it in the path:
C:\location_to_python\Lib\site-packages\unicorn\
3. Sample code
This directory contains some sample code to show how to use Unicorn API.
- sample_<arch>.py
These code show how to access architecture-specific information for each
architecture.
- shellcode.py
This shows how to analyze a Linux shellcode.
- sample_network_auditing.py
This shows how to analyze & interpret Linux shellcode.

42
bindings/python/README.md Normal file
View File

@ -0,0 +1,42 @@
# Python Bindings for Unicorn
## Install
Install a prebuilt wheel from PyPI:
```bash
pip3 install unicorn
```
In case you would like to develop the bindings:
```bash
# Python3
DEBUG=1 THREADS=4 pip3 install --user -e .
# Workaround for Pylance
DEBUG=1 THREADS=4 pip3 install --user -e . --config-settings editable_mode=strict
# Python2
DEBUG=1 THREADS=4 pip install -e .
```
or install it by building it by yourself:
```bash
# Python3
THREADS=4 pip3 install --user .
# Python2, unfortunately `pip2` doesn't support in-tree build
THREADS=4 python3 setup.py install
```
Explanations for arguments:
- `THREADS=4` will use 4 threads for building.
- `DEBUG=1` will build debug version of unicorn.
- `--user` will install the bindings to your user directory instead of requiring root permission.
- `-e` infers the editable mode, which gives your instant feedback instead of re-compiling every time.
Note that you should setup a valid building environment according to docs/COMPILE.md but not necessarily build it because `setup.py` will do this for you.
## Python2 compatibility
By default, Unicorn python bindings will be maintained against Python3 as it offers more powerful features which improves developing efficiency. Meanwhile, Unicorn will only keep compatible with all features Unicorn1 offers regarding Python2 because Python2 has reached end-of-life for more than 3 years as the time of writing this README. While offering all features for both Python2 & Python3 is desirable and doable, it inevitably costs too much efforts to maintain and few users really rely on this. Therefore, we assume that if users still stick to Python2, previous Unicorn1 features we offer should be enough. If you really want some new features Unicorn2 offers, please check and pull request to `unicorn/unicorn_py2``. We are happy to review and accept!

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,985 @@
# Unicorn Python bindings, by Nguyen Anh Quynnh <aquynh@gmail.com>
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 collections import namedtuple
# We can't place this file in a separate folder due to Python2 limitations but
# anyway we just maintain it with minimum efforts and it has been more than 3
# years since EOL of Python2 so it should be fine.
from . import x86_const, arm_const, arm64_const, unicorn_const as uc
if not hasattr(sys.modules[__name__], "__file__"):
__file__ = inspect.getfile(inspect.currentframe())
_python2 = sys.version_info[0] < 3
if _python2:
range = xrange
_lib = { 'darwin': 'libunicorn.2.dylib',
'win32': 'unicorn.dll',
'cygwin': 'cygunicorn.dll',
'linux': 'libunicorn.so.2',
'linux2': 'libunicorn.so.2' }
# Windows DLL in dependency order
_all_windows_dlls = (
"libwinpthread-1.dll",
"libgcc_s_seh-1.dll",
"libgcc_s_dw2-1.dll",
)
_loaded_windows_dlls = set()
def _load_win_support(path):
for dll in _all_windows_dlls:
if dll in _loaded_windows_dlls:
continue
lib_file = os.path.join(path, dll)
if ('/' not in path and '\\' not in path) or os.path.exists(lib_file):
try:
#print('Trying to load Windows library', lib_file)
ctypes.cdll.LoadLibrary(lib_file)
#print('SUCCESS')
_loaded_windows_dlls.add(dll)
except OSError as e:
#print('FAIL to load %s' %lib_file, e)
continue
# Initial attempt: load all dlls globally
if sys.platform in ('win32', 'cygwin'):
_load_win_support('')
def _load_lib(path):
try:
if sys.platform in ('win32', 'cygwin'):
_load_win_support(path)
lib_file = os.path.join(path, _lib.get(sys.platform, 'libunicorn.so.2'))
dll = ctypes.cdll.LoadLibrary(lib_file)
#print('SUCCESS')
return dll
except OSError as e:
#print('FAIL to load %s' %lib_file, e)
return None
_uc = None
# Loading attempts, in order
# - user-provided environment variable
# - pkg_resources can get us the path to the local libraries
# - we can get the path to the local libraries by parsing our filename
# - global load
# - python's lib directory
# - last-gasp attempt at some hardcoded paths on darwin and linux
_path_list = [os.getenv('LIBUNICORN_PATH', None),
pkg_resources.resource_filename(__name__, 'lib'),
os.path.join(os.path.split(__file__)[0], 'lib'),
'',
distutils.sysconfig.get_python_lib(),
"/usr/local/lib/" if sys.platform == 'darwin' else '/usr/lib64',
os.getenv('PATH', '')]
# print(_path_list)
#print("-" * 80)
for _path in _path_list:
if _path is None: continue
_uc = _load_lib(_path)
if _uc is not None: break
else:
raise ImportError("ERROR: fail to load the dynamic library.")
# __version__ = "%u.%u.%u" % (uc.UC_VERSION_MAJOR, uc.UC_VERSION_MINOR, uc.UC_VERSION_EXTRA)
# setup all the function prototype
def _setup_prototype(lib, fname, restype, *argtypes):
try:
getattr(lib, fname).restype = restype
getattr(lib, fname).argtypes = argtypes
except AttributeError:
raise ImportError("ERROR: Fail to setup some function prototypes. Make sure you have cleaned your unicorn1 installation.")
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
class _uc_mem_region(ctypes.Structure):
_fields_ = [
("begin", ctypes.c_uint64),
("end", ctypes.c_uint64),
("perms", ctypes.c_uint32),
]
class uc_tb(ctypes.Structure):
""""TranslationBlock"""
_fields_ = [
("pc", ctypes.c_uint64),
("icount", ctypes.c_uint16),
("size", ctypes.c_uint16)
]
_setup_prototype(_uc, "uc_version", ctypes.c_uint, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int))
_setup_prototype(_uc, "uc_arch_supported", ctypes.c_bool, ctypes.c_int)
_setup_prototype(_uc, "uc_open", ucerr, ctypes.c_uint, ctypes.c_uint, ctypes.POINTER(uc_engine))
_setup_prototype(_uc, "uc_close", ucerr, uc_engine)
_setup_prototype(_uc, "uc_strerror", ctypes.c_char_p, ucerr)
_setup_prototype(_uc, "uc_errno", ucerr, uc_engine)
_setup_prototype(_uc, "uc_reg_read", ucerr, uc_engine, ctypes.c_int, ctypes.c_void_p)
_setup_prototype(_uc, "uc_reg_write", ucerr, uc_engine, ctypes.c_int, ctypes.c_void_p)
_setup_prototype(_uc, "uc_mem_read", ucerr, uc_engine, ctypes.c_uint64, ctypes.POINTER(ctypes.c_char), ctypes.c_size_t)
_setup_prototype(_uc, "uc_mem_write", ucerr, uc_engine, ctypes.c_uint64, ctypes.POINTER(ctypes.c_char), ctypes.c_size_t)
_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)
_setup_prototype(_uc, "uc_mem_protect", ucerr, uc_engine, ctypes.c_uint64, ctypes.c_size_t, ctypes.c_uint32)
_setup_prototype(_uc, "uc_query", ucerr, uc_engine, ctypes.c_uint32, ctypes.POINTER(ctypes.c_size_t))
_setup_prototype(_uc, "uc_context_alloc", ucerr, uc_engine, ctypes.POINTER(uc_context))
_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))
# 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)
_setup_prototype(_uc, "uc_ctl", ucerr, uc_engine, ctypes.c_int)
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)
UC_HOOK_MEM_INVALID_CB = ctypes.CFUNCTYPE(
ctypes.c_bool, uc_engine, ctypes.c_int,
ctypes.c_uint64, ctypes.c_int, ctypes.c_int64, ctypes.c_void_p
)
UC_HOOK_MEM_ACCESS_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.c_int,
ctypes.c_uint64, ctypes.c_int, ctypes.c_int64, ctypes.c_void_p
)
UC_HOOK_INTR_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.c_uint32, ctypes.c_void_p
)
UC_HOOK_INSN_IN_CB = ctypes.CFUNCTYPE(
ctypes.c_uint32, uc_engine, ctypes.c_uint32, ctypes.c_int, ctypes.c_void_p
)
UC_HOOK_INSN_OUT_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.c_uint32,
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_HOOK_INSN_SYS_CB = ctypes.CFUNCTYPE(ctypes.c_uint32, uc_engine, ctypes.c_uint32, ctypes.c_void_p, 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
)
UC_HOOK_EDGE_GEN_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.POINTER(uc_tb), ctypes.POINTER(uc_tb), ctypes.c_void_p
)
UC_HOOK_TCG_OPCODE_CB = ctypes.CFUNCTYPE(
None, uc_engine, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_void_p
)
# access to error code via @errno of UcError
class UcError(Exception):
def __init__(self, errno):
self.errno = errno
def __str__(self):
return _uc.uc_strerror(self.errno).decode('ascii')
# return the core's version
def uc_version():
major = ctypes.c_int()
minor = ctypes.c_int()
combined = _uc.uc_version(ctypes.byref(major), ctypes.byref(minor))
return (major.value, minor.value, combined)
# return the binding's version
def version_bind():
return (
uc.UC_API_MAJOR, uc.UC_API_MINOR,
(uc.UC_API_MAJOR << 8) + uc.UC_API_MINOR,
)
# check to see if this engine supports a particular arch
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
if arch == uc.UC_ARCH_ARM:
if reg_id == arm_const.UC_ARM_REG_CP_REG:
reg = uc_arm_cp_reg()
if not isinstance(opt, tuple) or len(opt) != 7:
raise UcError(uc.UC_ERR_ARG)
reg.cp, reg.is64, reg.sec, reg.crn, reg.crm, reg.opc1, reg.opc2 = opt
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.val
if arch == uc.UC_ARCH_ARM64:
if reg_id == arm64_const.UC_ARM64_REG_CP_REG:
reg = uc_arm64_cp_reg()
if not isinstance(opt, tuple) or len(opt) != 5:
raise UcError(uc.UC_ERR_ARG)
reg.crn, reg.crm, reg.op0, reg.op1, reg.op2 = opt
status = reg_read_func(reg_id, ctypes.byref(reg))
if status != uc.UC_ERR_OK:
raise UcError(status)
return reg.val
elif 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)
# 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
def reg_write(reg_write_func, arch, reg_id, value):
reg = 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]:
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 arch == uc.UC_ARCH_ARM:
if reg_id == arm64_const.UC_ARM64_REG_CP_REG:
reg = uc_arm64_cp_reg()
if not isinstance(value, tuple) or len(value) != 6:
raise UcError(uc.UC_ERR_ARG)
reg.crn, reg.crm, reg.op0, reg.op1, reg.op2, reg.val = value
elif reg_id == arm_const.UC_ARM_REG_CP_REG:
reg = uc_arm_cp_reg()
if not isinstance(value, tuple) or len(value) != 8:
raise UcError(uc.UC_ERR_ARG)
reg.cp, reg.is64, reg.sec, reg.crn, reg.crm, reg.opc1, reg.opc2, reg.val = value
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
def _catch_hook_exception(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
"""Catches exceptions raised in hook functions.
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
self.emu_stop()
return wrapper
class uc_arm_cp_reg(ctypes.Structure):
"""ARM coprocessors registers for instructions MRC, MCR, MRRC, MCRR"""
_fields_ = [
("cp", ctypes.c_uint32),
("is64", ctypes.c_uint32),
("sec", ctypes.c_uint32),
("crn", ctypes.c_uint32),
("crm", ctypes.c_uint32),
("opc1", ctypes.c_uint32),
("opc2", ctypes.c_uint32),
("val", ctypes.c_uint64)
]
class uc_arm64_cp_reg(ctypes.Structure):
"""ARM64 coprocessors registers for instructions MRS, MSR"""
_fields_ = [
("crn", ctypes.c_uint32),
("crm", ctypes.c_uint32),
("op0", ctypes.c_uint32),
("op1", ctypes.c_uint32),
("op2", ctypes.c_uint32),
("val", ctypes.c_uint64)
]
class uc_x86_mmr(ctypes.Structure):
"""Memory-Management Register for instructions IDTR, GDTR, LDTR, TR."""
_fields_ = [
("selector", ctypes.c_uint16), # not used by GDTR and IDTR
("base", ctypes.c_uint64), # handle 32 or 64 bit CPUs
("limit", ctypes.c_uint32),
("flags", ctypes.c_uint32), # not used by GDTR and IDTR
]
class uc_x86_msr(ctypes.Structure):
_fields_ = [
("rid", ctypes.c_uint32),
("value", ctypes.c_uint64),
]
class uc_x86_float80(ctypes.Structure):
"""Float80"""
_fields_ = [
("mantissa", ctypes.c_uint64),
("exponent", ctypes.c_uint16),
]
class uc_x86_xmm(ctypes.Structure):
"""128-bit xmm register"""
_fields_ = [
("low_qword", ctypes.c_uint64),
("high_qword", ctypes.c_uint64),
]
class uc_x86_ymm(ctypes.Structure):
"""256-bit ymm register"""
_fields_ = [
("first_qword", ctypes.c_uint64),
("second_qword", ctypes.c_uint64),
("third_qword", ctypes.c_uint64),
("fourth_qword", ctypes.c_uint64),
]
class uc_arm64_neon128(ctypes.Structure):
"""128-bit neon register"""
_fields_ = [
("low_qword", ctypes.c_uint64),
("high_qword", ctypes.c_uint64),
]
# Subclassing ref to allow property assignment.
class UcRef(weakref.ref):
pass
# This class tracks Uc instance destruction and releases handles.
class UcCleanupManager(object):
def __init__(self):
self._refs = {}
def register(self, uc):
ref = UcRef(uc, self._finalizer)
ref._uch = uc._uch
ref._class = uc.__class__
self._refs[id(ref)] = ref
def _finalizer(self, ref):
# note: this method must be completely self-contained and cannot have any references
# to anything else in this module.
#
# This is because it may be called late in the Python interpreter's shutdown phase, at
# which point the module's variables may already have been deinitialized and set to None.
#
# Not respecting that can lead to errors such as:
# Exception AttributeError:
# "'NoneType' object has no attribute 'release_handle'"
# in <bound method UcCleanupManager._finalizer of
# <unicorn.unicorn.UcCleanupManager object at 0x7f0bb83e4310>> ignored
#
# For that reason, we do not try to access the `Uc` class directly here but instead use
# the saved `._class` reference.
del self._refs[id(ref)]
ref._class.release_handle(ref._uch)
class Uc(object):
_cleanup = UcCleanupManager()
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
raise UcError(uc.UC_ERR_VERSION)
self._arch, self._mode = arch, mode
self._uch = ctypes.c_void_p()
status = _uc.uc_open(arch, mode, ctypes.byref(self._uch))
if status != uc.UC_ERR_OK:
self._uch = None
raise UcError(status)
# internal mapping table to save callback & userdata
self._callbacks = {}
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):
if uch:
try:
status = _uc.uc_close(uch)
if status != uc.UC_ERR_OK:
raise UcError(status)
except: # _uc might be pulled from under our feet
pass
# emulate from @begin, and stop when reaching address @until
def emu_start(self, begin, until, timeout=0, count=0):
self._hook_exception = None
status = _uc.uc_emu_start(self._uch, begin, until, timeout, count)
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)
if status != uc.UC_ERR_OK:
raise UcError(status)
# return the value of a register
def reg_read(self, reg_id, opt=None):
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):
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):
return self.reg_read(x86_const.UC_X86_REG_MSR, msr_id)
# write to MSR - X86 only
def msr_write(self, msr_id, value):
return self.reg_write(x86_const.UC_X86_REG_MSR, (msr_id, value))
# read data from memory
def mem_read(self, address, size):
data = ctypes.create_string_buffer(size)
status = _uc.uc_mem_read(self._uch, address, data, size)
if status != uc.UC_ERR_OK:
raise UcError(status)
return bytearray(data)
# write to memory
def mem_write(self, address, data):
status = _uc.uc_mem_write(self._uch, address, data, len(data))
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)
if status != uc.UC_ERR_OK:
raise UcError(status)
# map a range of memory from a raw host memory address
def mem_map_ptr(self, address, size, perms, ptr):
status = _uc.uc_mem_map_ptr(self._uch, address, size, perms, ptr)
if status != uc.UC_ERR_OK:
raise UcError(status)
# unmap a range of memory
def mem_unmap(self, address, size):
status = _uc.uc_mem_unmap(self._uch, address, size)
if status != uc.UC_ERR_OK:
raise UcError(status)
# protect a range of memory
def mem_protect(self, address, size, perms=uc.UC_PROT_ALL):
status = _uc.uc_mem_protect(self._uch, address, size, perms)
if status != uc.UC_ERR_OK:
raise UcError(status)
# return CPU mode at runtime
def query(self, query_mode):
result = ctypes.c_size_t(0)
status = _uc.uc_query(self._uch, query_mode, ctypes.byref(result))
if status != uc.UC_ERR_OK:
raise UcError(status)
return result.value
@_catch_hook_exception
def _hook_tcg_op_cb(self, handle, address, arg1, arg2, user_data):
(cb, data) = self._callbacks[user_data]
cb(self, address, arg1, arg2, user_data)
@_catch_hook_exception
def _hook_edge_gen_cb(self, handle, cur, prev, user_data):
(cb, data) = self._callbacks[user_data]
cb(self, cur.contents, prev.contents, user_data)
@_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_sys_cb(self, handle, reg, pcp_reg, user_data):
cp_reg = ctypes.cast(pcp_reg, ctypes.POINTER(uc_arm64_cp_reg)).contents
uc_arm64_cp_reg_tuple = namedtuple("uc_arm64_cp_reg_tuple", ["crn", "crm", "op0", "op1", "op2", "val"])
(cb, data) = self._callbacks[user_data]
return cb(self, reg, uc_arm64_cp_reg_tuple(cp_reg.crn, cp_reg.crm, cp_reg.op0, cp_reg.op1, cp_reg.op2, cp_reg.val), 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]
cb(self, data)
def ctl(self, control, *args):
status = _uc.uc_ctl(self._uch, control, *args)
if status != uc.UC_ERR_OK:
raise UcError(status)
return status
def __ctl(self, ctl, nr, rw):
return ctl | (nr << 26) | (rw << 30)
def __ctl_r(self, ctl, nr):
return self.__ctl(ctl, nr, uc.UC_CTL_IO_READ)
def __ctl_w(self, ctl, nr):
return self.__ctl(ctl, nr, uc.UC_CTL_IO_WRITE)
def __ctl_rw(self, ctl, nr):
return self.__ctl(ctl, nr, uc.UC_CTL_IO_READ_WRITE)
def __ctl_r_1_arg(self, ctl, ctp):
arg = ctp()
self.ctl(self.__ctl_r(ctl, 1), ctypes.byref(arg))
return arg.value
def __ctl_w_1_arg(self, ctl, val, ctp):
arg = ctp(val)
self.ctl(self.__ctl_w(ctl, 1), arg)
def __ctl_rw_1_1_arg(self, ctl, val, ctp1, ctp2):
arg1 = ctp1(val)
arg2 = ctp2()
self.ctl(self.__ctl_rw(ctl, 2), arg1, ctypes.byref(arg2))
return arg2
def ctl_get_mode(self):
return self.__ctl_r_1_arg(uc.UC_CTL_UC_MODE, ctypes.c_int)
def ctl_get_page_size(self):
return self.__ctl_r_1_arg(uc.UC_CTL_UC_PAGE_SIZE, ctypes.c_uint32)
def ctl_set_page_size(self, val):
self.__ctl_w_1_arg(uc.UC_CTL_UC_PAGE_SIZE, val, ctypes.c_uint32)
def ctl_get_arch(self):
return self.__ctl_r_1_arg(uc.UC_CTL_UC_ARCH, ctypes.c_int)
def ctl_get_timeout(self):
return self.__ctl_r_1_arg(uc.UC_CTL_UC_TIMEOUT, ctypes.c_uint64)
def ctl_exits_enabled(self, val):
self.__ctl_w_1_arg(uc.UC_CTL_UC_USE_EXITS, val, ctypes.c_int)
def ctl_get_exits_cnt(self):
return self.__ctl_r_1_arg(uc.UC_CTL_UC_EXITS_CNT, ctypes.c_size_t)
def ctl_get_exits(self):
l = self.ctl_get_exits_cnt()
arr = (ctypes.c_uint64 * l)()
self.ctl(self.__ctl_r(uc.UC_CTL_UC_EXITS, 2), ctypes.cast(arr, ctypes.c_void_p), ctypes.c_size_t(l))
return [i for i in arr]
def ctl_set_exits(self, exits):
arr = (ctypes.c_uint64 * len(exits))()
for idx, exit in enumerate(exits):
arr[idx] = exit
self.ctl(self.__ctl_w(uc.UC_CTL_UC_EXITS, 2), ctypes.cast(arr, ctypes.c_void_p), ctypes.c_size_t(len(exits)))
def ctl_get_cpu_model(self):
return self.__ctl_r_1_arg(uc.UC_CTL_CPU_MODEL, ctypes.c_int)
def ctl_set_cpu_model(self, val):
self.__ctl_w_1_arg(uc.UC_CTL_CPU_MODEL, val, ctypes.c_int)
def ctl_remove_cache(self, addr):
self.__ctl_w_1_arg(uc.UC_CTL_TB_REMOVE_CACHE, addr, ctypes.c_uint64)
def ctl_request_cache(self, addr):
return self.__ctl_rw_1_1_arg(uc.UC_CTL_TB_REQUEST_CACHE, addr, ctypes.c_uint64, uc_tb)
# add a hook
def hook_add(self, htype, callback, user_data=None, begin=1, end=0, arg1=0, arg2=0):
_h2 = uc_hook_h()
# save callback & user_data
self._callback_count += 1
self._callbacks[self._callback_count] = (callback, user_data)
cb = None
if htype == uc.UC_HOOK_INSN:
insn = ctypes.c_int(arg1)
if arg1 == x86_const.UC_X86_INS_IN: # IN instruction
cb = ctypes.cast(UC_HOOK_INSN_IN_CB(self._hook_insn_in_cb), UC_HOOK_INSN_IN_CB)
if arg1 == x86_const.UC_X86_INS_OUT: # OUT instruction
cb = ctypes.cast(UC_HOOK_INSN_OUT_CB(self._hook_insn_out_cb), UC_HOOK_INSN_OUT_CB)
if arg1 in (x86_const.UC_X86_INS_SYSCALL, x86_const.UC_X86_INS_SYSENTER): # SYSCALL/SYSENTER instruction
cb = ctypes.cast(UC_HOOK_INSN_SYSCALL_CB(self._hook_insn_syscall_cb), UC_HOOK_INSN_SYSCALL_CB)
if arg1 in (arm64_const.UC_ARM64_INS_MRS, arm64_const.UC_ARM64_INS_MSR, arm64_const.UC_ARM64_INS_SYS, arm64_const.UC_ARM64_INS_SYSL):
cb = ctypes.cast(UC_HOOK_INSN_SYS_CB(self._hook_insn_sys_cb), UC_HOOK_INSN_SYS_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end), insn
)
elif htype == uc.UC_HOOK_TCG_OPCODE:
opcode = ctypes.c_int(arg1)
flags = ctypes.c_int(arg2)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, ctypes.cast(UC_HOOK_TCG_OPCODE_CB(self._hook_tcg_op_cb), UC_HOOK_TCG_OPCODE_CB),
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end), opcode, flags
)
elif htype == uc.UC_HOOK_INTR:
cb = ctypes.cast(UC_HOOK_INTR_CB(self._hook_intr_cb), UC_HOOK_INTR_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
elif htype == uc.UC_HOOK_INSN_INVALID:
cb = ctypes.cast(UC_HOOK_INSN_INVALID_CB(self._hook_insn_invalid_cb), UC_HOOK_INSN_INVALID_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
elif htype == uc.UC_HOOK_EDGE_GENERATED:
cb = ctypes.cast(UC_HOOK_EDGE_GEN_CB(self._hook_edge_gen_cb), UC_HOOK_EDGE_GEN_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
else:
if htype in (uc.UC_HOOK_BLOCK, uc.UC_HOOK_CODE):
# set callback with wrapper, so it can be called
# with this object as param
cb = ctypes.cast(UC_HOOK_CODE_CB(self._hookcode_cb), UC_HOOK_CODE_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
elif htype & (uc.UC_HOOK_MEM_READ_UNMAPPED |
uc.UC_HOOK_MEM_WRITE_UNMAPPED |
uc.UC_HOOK_MEM_FETCH_UNMAPPED |
uc.UC_HOOK_MEM_READ_PROT |
uc.UC_HOOK_MEM_WRITE_PROT |
uc.UC_HOOK_MEM_FETCH_PROT):
cb = ctypes.cast(UC_HOOK_MEM_INVALID_CB(self._hook_mem_invalid_cb), UC_HOOK_MEM_INVALID_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
else:
cb = ctypes.cast(UC_HOOK_MEM_ACCESS_CB(self._hook_mem_access_cb), UC_HOOK_MEM_ACCESS_CB)
status = _uc.uc_hook_add(
self._uch, ctypes.byref(_h2), htype, cb,
ctypes.cast(self._callback_count, ctypes.c_void_p),
ctypes.c_uint64(begin), ctypes.c_uint64(end)
)
# save the ctype function so gc will leave it alone.
self._ctype_cbs.append(cb)
if status != uc.UC_ERR_OK:
raise UcError(status)
return _h2.value
# delete a hook
def hook_del(self, h):
_h = uc_hook_h(h)
status = _uc.uc_hook_del(self._uch, _h)
if status != uc.UC_ERR_OK:
raise UcError(status)
h = 0
def context_save(self):
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)
return context
def context_update(self, context):
status = _uc.uc_context_save(self._uch, context.context)
if status != uc.UC_ERR_OK:
raise UcError(status)
def context_restore(self, context):
status = _uc.uc_context_restore(self._uch, context.context)
if status != uc.UC_ERR_OK:
raise UcError(status)
# this returns a generator of regions in the form (begin, end, perms)
def mem_regions(self):
regions = ctypes.POINTER(_uc_mem_region)()
count = ctypes.c_uint32()
status = _uc.uc_mem_regions(self._uch, ctypes.byref(regions), ctypes.byref(count))
if status != uc.UC_ERR_OK:
raise UcError(status)
try:
for i in range(count.value):
yield (regions[i].begin, regions[i].end, regions[i].perms)
finally:
_uc.uc_free(regions)
class UcContext:
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
@property
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, 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)
def __del__(self):
# We need this property since we shouldn't free it if the object is constructed from pickled bytes.
if self._to_free:
_uc.uc_context_free(self._context)
# print out debugging info
def debug():
archs = {
"arm": uc.UC_ARCH_ARM,
"arm64": uc.UC_ARCH_ARM64,
"mips": uc.UC_ARCH_MIPS,
"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 = ""
keys = archs.keys()
for k in sorted(keys):
if uc_arch_supported(archs[k]):
all_archs += "-%s" % k
major, minor, _combined = uc_version()
return "python-%s-c%u.%u-b%u.%u" % (
all_archs, major, minor, uc.UC_API_MAJOR, uc.UC_API_MINOR
)

View File

@ -0,0 +1 @@
from .unicorn import *

File diff suppressed because it is too large Load Diff