Unicorn 使用总结

Unicorn 是一个轻量级、跨平台的开源模拟器（Emulator），可以用于模拟不同架构的程序执行。它基于 QEMU 项目，但比 QEMU 更加轻量和易于使用，主要面向动态分析、漏洞挖掘和逆向工程等领域。Unicorn 允许开发者模拟目标平台的指令集架构（ISA），并且能够执行二进制代码，模拟程序运行时的行为，捕捉寄存器、内存等状态。

Unicorn 官方文档

Unicorn 安装

可以通过 Python 的包管理工具 pip 安装：

pip install unicorn

基本使用

创建 Unicorn 对象

和前面提到的 Keystone 和 Capstone 的流程类似，Unicorn 模拟执行首先需要创建一个 Unicorn 对象，并设置目标架构和模式。

from unicorn import *

# X86-64
from unicorn.x86_const import *
mu = Uc(UC_ARCH_X86, UC_MODE_64)
mu.mem_map(0, 0x1000) # 避免异常 "mov rax, gs:[0x28]"

# ARM
from unicorn.arm_const import *
mu = Uc(UC_ARCH_ARM, UC_MODE_THUMB)

# AARCH64
from unicorn.arm64_const import *
mu = Uc(UC_ARCH_ARM64, UC_MODE_ARM)

注意

在模拟 x86 架构时，程序可能会访问特定段（如 gs 或 fs）中的数据，这些数据通常与操作系统、线程信息、TLS（线程局部存储）等有关（例如 mov rax, gs:[0x28] 指令获取 canary）。

由于 Unicorn 模拟器仅提供模拟功能，导致像 fs、gs 这样的段寄存器都被默认初始化为 0，因此访问这些段中数据的指令会触发程序内存读写异常。

为了避免这种情况影响程序模拟执行，最好在 0 地址处映射一段内存。

mu.mem_map(0, 0x1000) # 避免异常 "mov rax, gs:[0x28]"

寄存器操作

Unicorn 允许你模拟寄存器的读写操作，你可以通过 mu.reg_write() 来写入寄存器的值，使用 mu.reg_read() 来读取寄存器的值。

写入寄存器

mu.reg_write(reg_id, value) 用来写入寄存器的值，reg_id 是寄存器的标识符，value 是要写入的值。

注意

一些特殊的寄存器如 gs 寄存器不能直接通过 reg_write 修改。

mu.reg_write(UC_X86_REG_RIP, 0x1000)  # 设置 RIP 寄存器的值

读取寄存器

mu.reg_read(reg_id) 返回指定寄存器的当前值。

rip_value = mu.reg_read(UC_X86_REG_RIP)  # 读取 RIP 寄存器的值
print(f"RIP: {rip_value}")

内存操作

Unicorn 允许你模拟内存的读写操作，模拟程序运行时的内存访问。你可以通过 mu.mem_map() 来映射内存区域，通过 mu.mem_write() 来写入数据，通过 mu.mem_read() 来读取内存数据。

映射内存

mu.mem_map(start, size) 用来映射一段内存区域，start 是内存起始地址，size 是内存的大小。

注意

• start 和 end 要关于 0x1000 对齐。
• 要确保映射的内存地址空间不要与之前映射的内存出现重叠。

mu.mem_map(0x1000, 0x1000)  # 映射 0x1000 大小的内存

写入内存

mu.mem_write(addr, data) 将数据写入指定地址的内存区域。data 是二进制数据（字节串），addr 是写入的地址。

code = b"\x48\x31\xc0"  # x86_64: xor rax, rax
mu.mem_write(0x1000, code)  # 将机器码写入模拟内存

读取内存

mu.mem_read(addr, size) 从指定地址读取 size 字节的数据。

data = mu.mem_read(0x1000, 4)  # 从地址 0x1000 读取 4 字节
print(data)

加载二进制程序

实际情况下我们可能需要模拟执行一个二进制程序中的某个解密算法函数。与 shellcode 不同，在二进制程序中即使一个纯算法函数也不是只加载代码段就可以正常仿真的，因为这个函数在执行过程中还可能访问全局变量。因此我们需要想办法将这个二进制程序加载到内存中，因此通常 Unicorn 还要配合 angr 的静态分析功能使用。

下面这段代码是加载二进制到 Unicorn 中的模板，可以应对大多数情况。

提示

• 这个模板的段映射部分写的很奇怪，这是因为有的二进制程序（比如一些 Linux 内核镜像）的段并不是关于 0x1000 对齐，但是 Unicorn 的 mem_map 要求地址范围按 0x1000 对齐，并且不能与之前映射的内存重叠。因此这里要做内存对齐处理并且多次捕获异常。
• x86 架构设置 bp 寄存器是因为有的函数会通过 bp 寄存器访问函数的参数和局部变量。
• 为了实现代码复用，通常的做法是把每个架构相关的操作抽象为接口类，然后每个架构实现这个接口类，而真正的核心逻辑不需要关注架构的细节。下面的模板就是遵循这一设计原则。

# 定义 amd64 架构的上下文
class amd64_ctx:
    # 架构寄存器映射
    regs = {
        'sp': UC_X86_REG_RSP,  # 堆栈指针寄存器（RSP）
        'bp': UC_X86_REG_RBP,  # 基指针寄存器（RBP）
    }

    def get_emu(self):
        # 创建并返回一个 X86-64 架构的模拟器对象
        emu = Uc(UC_ARCH_X86, UC_MODE_64)
        emu.mem_map(0, 0x1000)  # 映射内存，以避免访问 gs 或 fs 时发生异常
        return emu

# 定义 ARM 架构的上下文
class arm_ctx:
    # 架构寄存器映射
    regs = {
        'sp': UC_ARM_REG_SP,  # 堆栈指针寄存器（SP）
    }

    def get_emu(self):
        # 创建并返回一个 ARM 架构的模拟器对象（使用 THUMB 模式）
        return Uc(UC_ARCH_ARM, UC_MODE_THUMB)

# 定义 AARCH64 架构的上下文
class aarch64_ctx:
    # 架构寄存器映射
    regs = {
        'sp': UC_ARM64_REG_SP,  # 堆栈指针寄存器（SP）
    }

    def get_emu(self):
        # 创建并返回一个 AARCH64 架构的模拟器对象（使用 ARM 模式）
        return Uc(UC_ARCH_ARM64, UC_MODE_ARM)

# 初始化 angr 项目，加载目标二进制文件，禁用自动加载库
project = angr.Project(binary, load_options={'auto_load_libs': False})

# 根据架构类型选择对应的上下文对象
if project.arch.name == "ARMEL":
    arch_ctx = arm_ctx()
elif project.arch.name == "AMD64":
    arch_ctx = amd64_ctx()
elif project.arch.name == "AARCH64":
    arch_ctx = aarch64_ctx()
else:
    # 如果不支持的架构，抛出异常
    raise Exception("Unsupported architecture")

# 获取模拟器对象
emu = arch_ctx.get_emu()

# 设置堆栈地址和大小
STACK_ADDRESS = 0x1234000
STACK_SIZE = 0x3000

# 为堆栈区域映射内存
emu.mem_map(STACK_ADDRESS, STACK_SIZE)

# 设置堆栈指针（SP）寄存器
emu.reg_write(arch_ctx.regs['sp'], STACK_ADDRESS + STACK_SIZE - 0x1000)

# 如果有基指针（BP）寄存器，设置 BP 寄存器
if 'bp' in arch_ctx.regs:
    emu.reg_write(arch_ctx.regs['bp'], STACK_ADDRESS + STACK_SIZE - 0x1000 + 0x300)

# 映射二进制文件的各个段到模拟器内存
for seg in project.loader.main_object.segments:
    # 计算段的起始地址和结束地址（对齐到页面边界）
    mem_start = seg.vaddr & ~0xFFF
    mem_end = (seg.vaddr + seg.memsize + 0xFFF) & ~0xFFF
    
    try:
        # 映射内存区域
        emu.mem_map(mem_start, mem_end - mem_start)
    except:
        # 如果映射失败，尝试重新计算起始地址
        mem_start = (seg.vaddr + 0xFFF) & ~0xFFF
        if mem_start < mem_end and mem_start:
            emu.mem_map(mem_start, mem_end - mem_start)
        else:
            continue
    
    # 将段的内容写入模拟器内存
    emu.mem_write(seg.vaddr, project.loader.memory.load(seg.vaddr, seg.filesize))

模拟执行

虚拟机启动

当前期虚拟机初始化完成之后（主要是程序加载、栈初始化、寄存器设置等），就可以启动虚拟机进行模拟了。虚拟机通过 emu_start 函数启动，该函数定义如下：

def emu_start(self, begin: int, until: int, timeout: int=0, count: int=0) -> None:

• begin ：仿真开始的地址。表示从该地址开始执行机器码，通常是程序的入口点。
• until ：仿真结束的地址。表示程序执行到该地址时仿真停止。until 应大于 begin，否则可能导致仿真无限执行。
• timeout ：超时设置（单位：毫秒）。在仿真执行过程中，如果超时值非零，仿真会在超时后自动停止。如果设置为 0，表示不限制时间。
• count ：执行次数。如果 count 非零，仿真会执行指定的次数后停止。如果为 0，表示按地址范围执行，直到 end 地址或其他条件触发。

在虚拟机执行过程中可能会抛出异常，如访问非法内存地址、执行未定义指令等，通常这些异常会触发钩子函数或返回错误信息。为了应对这些可能的异常，可以通过在模拟中添加错误处理代码，保证仿真顺利进行或在发生错误时及时停止仿真。

常见的异常包括：

• UC_ERR_READ_UNMAPPED ：尝试读取未映射的内存。
• UC_ERR_WRITE_UNMAPPED ：尝试写入未映射的内存。
• UC_ERR_FETCH_UNMAPPED ：尝试获取未映射的指令。
• UC_ERR_READ_PROT ：尝试读取保护内存。
• UC_ERR_WRITE_PROT ：尝试写入保护内存。
• UC_ERR_FETCH_PROT ：尝试获取受保护的指令。

异常处理可以通过 try-except 结构来完成。下面是一个例子，展示了如何捕捉异常并处理：

try:
    # 启动仿真，从地址 0x1000 开始执行，直到 0x1003
    mu.emu_start(0x1000, 0x1003)
except UcError as e:
    print(f"Error during emulation: {e}")

事件监控

Unicorn 提供了事件钩子（hooks）功能，用于在程序执行过程中捕获特定的事件（如指令执行、内存读写、寄存器访问等）。通过钩子机制，用户可以在仿真时监控程序的特定行为并调用用户自己实现的回调函数处理。

Unicorn 提供了多种类型的监控事件，这里只介绍我们常用的事件类型。

注意

在添加事件钩子之后 Unicorn 的执行效率会变得非常低。因此在实际编写脚本的时候应当尽量避免使用 Unicorn 的事件监控功能，而是将其作为调试脚本的工具。

例如在一段代码中间的某个位置处需要做一些额外操作，时候可以可以以该位置为边界将代码分为两部分模拟，而不是设置一个指令执行事件来监控程序只不是执行到需要做额外操作的位置了。

指令执行事件

指令执行事件 UC_HOOK_CODE 在每执行一条指令时触发，该事件对应的回调函数如下：

def hook_code(uc, address, size, user_data):

• uc ：当前的 Unicorn 仿真对象，类型为 Uc。它是对当前仿真状态的一个引用，用户可以通过它访问模拟环境中的寄存器、内存等。
• address ：当前执行指令的地址（内存中的位置）。这个地址是指指令被执行时在仿真内存中加载的地址。
• size ：当前指令的大小（以字节为单位）。这表示指令在内存中占用的字节数。
• user_data ：用户自定义的数据。可以在调用 hook_add 时传入，通常用于传递上下文信息或者保持状态。

我们通常使用下面这个模板打印模拟执行的汇编指令来调试脚本：

from capstone import *
cs = Cs(CS_ARCH_X86, CS_MODE_64)

# 定义一个回调函数，用于监控指令执行
def hook_code(uc, address, size, user_data):
    insn = next(cs.disasm(uc.mem_read(address, size), address))
    print(f"{address:#x}: {insn.mnemonic} {insn.op_str}")

mu.hook_add(UC_HOOK_CODE, hook_code)  # 注册指令执行钩子

内存事件

我们常用的内存事件有内存读入事件（UC_HOOK_MEM_READ）和内存写入事件（UC_HOOK_MEM_WRITE），另外针对指令的有获取指令事件（UC_HOOK_MEM_FETCH）。对应的回调函数定义如下：

def hook_mem(uc, access, address, size, value, user_data):

• uc ：当前的 Unicorn 仿真对象，类型为 Uc，同样提供对仿真环境的访问。
• access ：内存访问的类型。可以是以下几种类型：
  • UC_MEM_READ：表示是数据读取。
  • UC_MEM_WRITE：表示是数据写入。
  • UC_MEM_FETCH：表示读取的是指令（通常用于指令获取）。
• address ：发生内存访问的地址（内存中的位置）。这个地址指的是访问的内存单元的地址。
• size ：内存读取操作的大小（以字节为单位）。表示读取或写入了多少字节的数据。
• value ：读取或写入的值。指在指定地址处读取或写入的实际内容（以整数形式表示）。
• user_data ：用户自定义的数据，通常用于在钩子中传递一些额外的上下文信息。

# 定义一个回调函数，用于监控内存读取
def hook_mem_read(uc, access, address, size, value, user_data):
    print(f"Memory read at address: 0x{address:X}, size: {size}, value: 0x{value:X}")

# 定义一个回调函数，用于监控内存写入
def hook_mem_write(uc, access, address, size, value, user_data):
    print(f"Memory write at address: 0x{address:X}, size: {size}, value: 0x{value:X}")

# 定义一个回调函数，用于监控内存获取指令
def hook_mem_fetch(uc, address, size, user_data):
    print(f"Instruction fetched from address: 0x{address:X}, size: {size} bytes")
    
mu.hook_add(UC_HOOK_MEM_READ, hook_mem_read)  # 注册内存读取钩子
mu.hook_add(UC_HOOK_MEM_WRITE, hook_mem_write)  # 注册内存写入钩子
mu.hook_add(UC_HOOK_MEM_FETCH, hook_mem_fetch)  # 注册内存获取指令钩子

常用脚本

trace shellcode

import time

from unicorn import *
from unicorn.x86_const import *
from capstone import *

CODE_ADDRESS = 0x74B000  # 0x6F8618  # Shellcode 的加载地址
STACK_ADDRESS = 0x0019C000  # 栈的起始地址（分配在高地址区域）
STACK_SIZE = 0x00080000  # 栈大小
ORIG_SP = STACK_ADDRESS + STACK_SIZE // 2
JUMP_BLOCK_RANGE = range(CODE_ADDRESS + 0x3fd0, CODE_ADDRESS + 0x137f0)

shellcode = open("shellcode.bin", "rb").read()

# 初始化 Unicorn 模拟器
mu = Uc(UC_ARCH_X86, UC_MODE_32)

# 映射内存
mu.mem_map(CODE_ADDRESS & ~0xFFF, ((len(shellcode) + 0xFFF) & ~0xFFF) + 0x1000)  # 为代码分配内存

mu.mem_map(STACK_ADDRESS, STACK_SIZE)  # 为栈分配内存

# 写入 Shellcode
mu.mem_write(CODE_ADDRESS, shellcode)

# 初始化栈指针（ESP 指向栈顶）
mu.reg_write(UC_X86_REG_ESP, STACK_ADDRESS + 0x0019F998 - 0x0019C000)

# 初始化 Capstone 用于反汇编
cs = Cs(CS_ARCH_X86, CS_MODE_32)


f = open("trace.txt", "w+")


class TraceInfo:
    def __init__(self, uc: unicorn.Uc, address, size):
        code = uc.mem_read(address, size)
        self.addr = address
        self.insn = next(cs.disasm(code, address))
        self.regs = self.get_registers(uc)
        self.regs_change = ""
        self.mem_change = ""

    def get_registers(self, uc: unicorn.Uc):
        return {
            'eax': uc.reg_read(UC_X86_REG_EAX),
            'ebx': uc.reg_read(UC_X86_REG_EBX),
            'ecx': uc.reg_read(UC_X86_REG_ECX),
            'edx': uc.reg_read(UC_X86_REG_EDX),
            'esi': uc.reg_read(UC_X86_REG_ESI),
            'edi': uc.reg_read(UC_X86_REG_EDI),
            'esp': uc.reg_read(UC_X86_REG_ESP),
            'ebp': uc.reg_read(UC_X86_REG_EBP),
            # 'eip': uc.reg_read(UC_X86_REG_EIP),
            'eflags': uc.reg_read(UC_X86_REG_EFLAGS),

            # 段寄存器（如需要可以添加调试信息）
            # 'cs': uc.reg_read(UC_X86_REG_CS),
            # 'ds': uc.reg_read(UC_X86_REG_DS),
            # 'es': uc.reg_read(UC_X86_REG_ES),
            # 'fs': uc.reg_read(UC_X86_REG_FS),
            # 'gs': uc.reg_read(UC_X86_REG_GS),
            # 'ss': uc.reg_read(UC_X86_REG_SS),
        }

    def set_regs_change(self, uc: unicorn.Uc):
        new_regs = self.get_registers(uc)
        diffs = []
        for reg, old_val in self.regs.items():
            new_val = new_regs[reg]
            if old_val != new_val:
                diffs.append(f"{reg}: {hex(old_val)}-> {hex(new_val)}")
        self.regs_change = " ".join(diffs)

    def set_mem_change(self, uc: unicorn.Uc, address, size, value):
        old_value = int.from_bytes(uc.mem_read(address, size), byteorder="little")
        self.mem_change = f"{hex(address)}: {hex(old_value)}-> {hex(value)}"

    def get_state_trace_info(self):
        addr_str = f"{hex(self.addr)}".ljust(10)
        insn_str = f"{self.insn.mnemonic} {self.insn.op_str}".ljust(40)
        regs_str = self.regs_change.ljust(60)
        mem_str = self.mem_change.ljust(60)
        return f"{addr_str} | {insn_str} | {regs_str} | {mem_str}"


trace_info: TraceInfo = None


# Hook：捕获内存写入
def hook_mem_write(uc, access, address, size, value, user_data):
    trace_info.set_mem_change(uc, address, size, value)

# 初始化计数器和时间
trace_count = 0
start_time = time.time()

# Hook：捕获指令执行后的状态
def hook_code(uc, address, size, user_data):
    global trace_info, trace_count, start_time

    # 更新指令计数器
    trace_count += 1

    # 动态显示执行速度
    current_time = time.time()
    elapsed_time = current_time - start_time
    if elapsed_time > 1:
        print(
            f"\rProcessed: {trace_count} instructions, Speed: {trace_count / elapsed_time:.2f} instructions/sec",
            end="",
        )

    if trace_info is not None:
        trace_info.set_regs_change(uc)
        print(trace_info.get_state_trace_info(), file=f)

    trace_info = TraceInfo(uc, address, size)


# 添加 Hook
mu.hook_add(UC_HOOK_MEM_WRITE, hook_mem_write)  # 捕获内存写入
mu.hook_add(UC_HOOK_CODE, hook_code)  # 捕获指令执行后的状态

# 开始模拟执行
try:
    print("Starting execution...")
    mu.emu_start(CODE_ADDRESS, CODE_ADDRESS + len(shellcode))
    print("Execution finished.")
except UcError as e:
    print(f"Unicorn execution failed: {e}")
    print(hex(mu.reg_read(UC_X86_REG_ESP)))
    print(hex(mu.reg_read(UC_X86_REG_EIP)))
    esp = mu.reg_read(UC_X86_REG_ESP)
    for i in range(esp, esp + 0x50, 4):
        print(hex(int.from_bytes(mu.mem_read(i, 4), byteorder="little")))