angr 0ctf_trace 模拟执行MIPS

发布于 2019-05-16  98 次阅读

分析

题目给了一个log和一个纯机器码文件
机器码文件是MIPS架构的，我们可以用32位IDA打开，然后选择目标处理器为小端的MIPS即可。
MIPS指令
通过分析，我们发现在栈上有一段缓冲区最先由已知字符串常量初始化的，然后一个未知的字符串（也就是flag）被添加到后面，最终由相同的快速排序转化进行了排序。我们可以通过符号执行，将状态一步步转移到已知分支，然后符号的值就是flag了。
代码分析https://github.com/fangdada/ctf/tree/master/0CTF2016/trace

反汇编

也可以通过如下脚本反汇编机器码

#!/usr/bin/env python

from __future__ import print_function
from capstone import *
from capstone.mips import *

f=open('./data.bin','rb')

arch=CS_ARCH_MIPS
mode=CS_MODE_MIPS32+CS_MODE_LITTLE_ENDIAN
code=f.read()
comment="MIPS-32 (little-endian)"

print("platform:"+comment)
print("disasm:")

try:
    md=Cs(arch,mode)
    md.detial=True
    for insn in md.disasm(code,0):
        print("0x%x: \t%s\t%s" % (insn.address, insn.mnemonic, insn.op_str))

except CsError as e:
    print("ERROR: %s" % e)

对应LOG文件和反汇编信息，可以得知，程序的加载地址为0x004009d4

bin文件加载

我们要用执行引擎模拟执行，先来看如下代码：

project = angr.Project("./data.bin", load_options={
    'main_opts': {
        'backend': 'blob',
        'base_addr': 0x400770,
        'arch': 'mipsel',
    },
})

跟常规的加载选项不同，这一次用了blob后端引擎使用mipsel架构来加载机器码文件，然后就可以模拟执行了。

Trace

state = project.factory.blank_state(addr=MAIN_START)
state.memory.store(FLAG_LOCATION, state.solver.BVS("flag", 8*32))
state.memory.store(FLAG_PTR_LOCATION, struct.pack("<I", FLAG_LOCATION))

sm = project.factory.simulation_manager(state)
choices = [state]

因为是用机器码模拟执行所以要用空状态初始化，然后设置一下开始地址，再把solver的变量存储进入求解，最后同样作为参数传给simulation_manager就可以准备执行了。

  • 解析trace的log文件代码如下：
def load_trace():
    res = []
    delay_slots = set()
    with open("./trace_8339a701aae26588966ad9efa0815a0a.log") as f:
        for line in f:
            if line.startswith('[INFO]'):
                addr = int(line[6:6+8], 16)

                res.append(addr)

                # every command like this is in delay slot
                # (in this particular binary)
                if ("move r1, r1" in line):
                    delay_slots.add(addr)

    return res, delay_slots

trace_log保存了trace的每个指令的地址，delay_slots保存了适应处理器流水线设计的延迟指令。

trace_log, delay_slots = load_trace()

print("Tracing...")
    for i, addr in enumerate(trace_log):
        if addr in delay_slots:
            continue

        for s in choices:
            if s.addr == addr:
                break

        else:
            raise ValueError("couldn't advance to %08x, line %d" % (addr, i+1))

        if s.addr == MAIN_END:
            break

        # if command is a jump, it's followed by a delay slot
        # we need to advance by two instructions
        # https://github.com/angr/angr/issues/71
        if s.addr + 4 in delay_slots:
            choices = project.factory.successors(s, num_inst=2).successors
        else:
            choices = project.factory.successors(s, num_inst=1).successors

    state = s

重点来看choices=project.factory.successors(s, num_inst=2).successors。
首先我们来看看官方文档里的解释：

successors(simgr, state, **kwargs)

Perform the process of stepping a state forward, returning a SimSuccessors object.

  • num_inst – The maximum number of instructions.

successors会使用适合的模拟器（Simulation）执行，步进（step）一个状态（state），num_inst指定了最大的步进的语句数。num_inst=2就是为了跳过延迟指令delay slots，否则会丢失状态分支，原因正如官方注释里的那个issue。之前那个for s in choices……的语句段就是让s指向最后的状态，而不会让s指向那个delay slot，不然脚本会出问题。

等脚本最终trace完了之后state=s使state为最终状态，这时候之前保存的符号变量应当为flag了，将solver里的值取出转化为字符串就能得到flag了：

print("Running solver...")
solution = state.solver.eval(state.memory.load(FLAG_LOCATION, 32), cast_to=bytes)
print("The flag is", bytes.decode(solution).strip('\x00'))
return solution

EXP

#!/usr/bin/env python2

"""
In this challenge we're given a text file with trace of a program execution. The file has
two columns, address and instruction executed. So we know all the instructions being executed,
and which branches were taken. But the initial data is not known.
Reversing reveals that a buffer on the stack is initialized with known constant string first,
then an unknown string is appended to it (the flag), and finally it's sorted with some
variant of quicksort. And we need to find the flag somehow.
angr easily solves this problem. We only have to direct it to the right direction
at every branch, and solver finds the flag at a glance.
"""

from __future__ import print_function

import struct

import angr

MAIN_START = 0x4009d4
MAIN_END = 0x00400c18

FLAG_LOCATION = 0x400D80
FLAG_PTR_LOCATION = 0x410EA0


def load_trace():
    res = []
    delay_slots = set()
    with open("./trace_8339a701aae26588966ad9efa0815a0a.log") as f:
        for line in f:
            if line.startswith('[INFO]'):
                addr = int(line[6:6 + 8], 16)

                res.append(addr)

                # every command like this is in delay slot
                # (in this particular binary)
                if ("move r1, r1" in line):
                    delay_slots.add(addr)

    return res, delay_slots


def main():
    trace_log, delay_slots = load_trace()

    # data.bin is simply the binary assembled from trace,
    # starting on 0x400770
    project = angr.Project("./data.bin", load_options={
        'main_opts': {
            'backend': 'blob',
            'base_addr': 0x400770,
            'arch': 'mipsel',
        },
    })

    state = project.factory.blank_state(addr=MAIN_START,add_options=angr.options.unicorn)
    state.memory.store(FLAG_LOCATION, state.solver.BVS("flag", 8 * 32))
    state.memory.store(FLAG_PTR_LOCATION, struct.pack("<I", FLAG_LOCATION))

    # sm = project.factory.simulation_manager(state) #why? not use it
    choices = [state]

    print("Tracing...")
    for i, addr in enumerate(trace_log):
        if addr in delay_slots:
            continue

        for s in choices:  # find the state of this address in the choices
            if s.addr == addr:
                break

        else:
            raise ValueError("couldn't advance to %08x, line %d" % (addr, i + 1))

        if s.addr == MAIN_END:
            break

        # if command is a jump, it's followed by a delay slot
        # we need to advance by two instructions
        # https://github.com/angr/angr/issues/71
        if s.addr + 4 in delay_slots:
            choices = project.factory.successors(s, num_inst=2).successors
        else:
            choices = project.factory.successors(s, num_inst=1).successors

    state = s

    print("Running solver...")

    solution = state.solver.eval(state.memory.load(FLAG_LOCATION, 32), cast_to=bytes)
    print("The flag is", bytes.decode(solution).strip('\x00'))

    return solution


# def test():
#     assert main() == "0ctf{tr135m1k5l96551s9l5r}"


if __name__ == "__main__":
    main()

知识点