Rex: 自动化利用引擎分析

docker pull shellphish/mechaphish; 
docker run -it shellphish/mechaphish

tg = archr.targets.LocalTarget(<path_to_binary>, target_os='cgc')
crash = rex.Crash(tg, <input>)

t = archr.targets.LocalTarget(["/home/angr-dev/binaries/tests/defcon24/legit_00003"], target_os='cgc')
crash = rex.Crash(t, b"\x00\x0b1\xc1\x00\x0c\xeb\xe4\xf1\xf1\x14\r\rM\r\xf3\x1b\r\r\r~\x7f\x1b\xe3\x0c`_222\r\rM\r\xf3\x1b\r\x7f\x002\x7f~\x7f\xe2\xff\x7f\xff\xff\x8b\xc7\xc9\x83\x8b\x0c\xeb\x80\x002\xac\xe2\xff\xff\x00t\x8bt\x8bt_o_\x00t\x8b\xc7\xdd\x83\xc2t~n~~\xac\xe2\xff\xff_k_\x00t\x8b\xc7\xdd\x83\xc2t~n~~\xac\xe2\xff\xff\x00t\x8bt\x8b\xac\xf1\x83\xc2t~c\x00\x00\x00~~\x7f\xe2\xff\xff\x00t\x9e\xac\xe2\xf1\xf2@\x83\xc3t")

IP_OVERWRITE              = "ip_overwrite"
    PARTIAL_IP_OVERWRITE      = "partial_ip_overwrite"
    UNCONTROLLED_IP_OVERWRITE = "uncontrolled_ip_overwrite"
    BP_OVERWRITE              = "bp_overwrite"
    PARTIAL_BP_OVERWRITE      = "partial_bp_overwrite"
    WRITE_WHAT_WHERE          = "write_what_where"
    WRITE_X_WHERE             = "write_x_where"
    UNCONTROLLED_WRITE        = "uncontrolled_write" # a write where the destination address is uncontrolled
    ARBITRARY_READ            = "arbitrary_read"
    NULL_DEREFERENCE          = "null_dereference"
    ARBITRARY_TRANSMIT        = "arbitrary_transmit" # transmit where the buf argument is completely controlled
    ARBITRARY_RECEIVE         = "arbitrary_receive" # receive where the buf argument is completel controlled

# collect a concrete trace
   save_core = True
   if isinstance(self.tracer_bow, archr.arsenal.RRTracerBow):
       save_core = False
   r = self.tracer_bow.fire(testcase=test_case, channel=channel,save_core=save_core)

self._t = r.tracer_technique(keep_predecessors=2, copy_states=False, mode=TracingMode.Strict)
simgr.use_technique(self._t)
simgr.use_technique(angr.exploration_techniques.Oppologist())

# tracing completed
        # if there was no crash we'll have to use the previous path's state
        if 'crashed' in simgr.stashes:
            # the state at crash time
            self.state = simgr.crashed[0]
            # a path leading up to the crashing basic block
            self.prev = self._t.predecessors[-1]
        else:
            self.state = simgr.traced[0]
            self.prev = self.state

def _filter_memory_writes(self):
        memory_writes = sorted(self.state.memory.mem.get_symbolic_addrs())
        if self.is_cgc:
            # remove all memory writes that directly end up in the CGC flag page (0x4347c000 - 0x4347d000)
            memory_writes = [m for m in memory_writes if m // 0x1000 != 0x4347c]
        user_writes = [m for m in memory_writes if
                       any("aeg_stdin" in v for v in self.state.memory.load(m, 1).variables)]
        if self.is_cgc:
            flag_writes = [m for m in memory_writes if
                           any(v.startswith("cgc-flag") for v in self.state.memory.load(m, 1).variables)]
        else:
            flag_writes = []
        l.debug("Finished filtering memory writes.")
        self.symbolic_mem = self._segment(user_writes)
        self.flag_mem = self._segment(flag_writes)

# 判断 ip 是否可控，bp 类似        
    if self.state.solver.symbolic(ip): 
        # how much control of ip do we have?
        if self._symbolic_control(ip) >= self.state.arch.bits:
            l.info("detected ip overwrite vulnerability")
            self.crash_types.append(Vulnerability.IP_OVERWRITE)
        else:
            l.info("detected partial ip overwrite vulnerability")
            self.crash_types.append(Vulnerability.PARTIAL_IP_OVERWRITE)

        return

# 筛选出目的地址可控的操作
        # grab the all actions in the last basic block
        symbolic_actions = [ ]
        if self._t is not None and self._t.last_state is not None:
            recent_actions = reversed(self._t.last_state.history.recent_actions)
            state = self._t.last_state
            # TODO: this is a dead assignment! what was this supposed to be?
        else:
            recent_actions = reversed(self.state.history.actions)
            state = self.state
        for a in recent_actions:
            if a.type == 'mem':
                if self.state.solver.symbolic(a.addr.ast):
                    symbolic_actions.append(a)

#判断是内存读还是内存写，并判断数据是否可控，由此确定漏洞类型
        for sym_action in symbolic_actions:
            if sym_action.action == "write":
                if self.state.solver.symbolic(sym_action.data):
                    l.info("detected write-what-where vulnerability")
                    self.crash_types.append(Vulnerability.WRITE_WHAT_WHERE)
                else:
                    l.info("detected write-x-where vulnerability")
                    self.crash_types.append(Vulnerability.WRITE_X_WHERE)

                self.violating_action = sym_action
                break

            if sym_action.action == "read":
                # special vulnerability type, if this is detected we can explore the crash further
                l.info("detected arbitrary-read vulnerability")
                self.crash_types.append(Vulnerability.ARBITRARY_READ)

                self.violating_action = sym_action
                break

def explorable(self):
        explorables = [Vulnerability.ARBITRARY_READ, Vulnerability.WRITE_WHAT_WHERE, Vulnerability.WRITE_X_WHERE]
        return self.one_of(explorables)

if self.one_of([Vulnerability.ARBITRARY_READ]):
                self._explore_arbitrary_read(path_file)
        elif self.one_of([Vulnerability.WRITE_WHAT_WHERE, Vulnerability.WRITE_X_WHERE]):
            self._explore_arbitrary_write(path_file)
        else:
            raise CannotExplore("unknown explorable crash type: %s" % self.crash_types)

def exploitable(self):
        exploitables = [Vulnerability.IP_OVERWRITE, Vulnerability.PARTIAL_IP_OVERWRITE, Vulnerability.BP_OVERWRITE,
                Vulnerability.PARTIAL_BP_OVERWRITE, Vulnerability.WRITE_WHAT_WHERE, Vulnerability.WRITE_X_WHERE]
        return self.one_of(exploitables)

def exploit(self, blacklist_symbolic_explore=True, **kwargs):
        """
        Initialize an exploit factory, with which you can build exploits.
        :return:    An initialized ExploitFactory instance.
        :rtype:     ExploitFactory
        """
        factory = self._prepare_exploit_factory(blacklist_symbolic_explore, **kwargs)
        factory.initialize()
        return factory

def test_legit_00003():
    # Test exploration and exploitation of legit_00003.
    inp = b"1\n" + b"A" * 200 #设置输入内容
    path = os.path.join(bin_location, "tests/defcon24/legit_00003")
    with archr.targets.LocalTarget([path], target_os='cgc') as target:
        crash = rex.Crash(target, inp, fast_mode=True, rop_cache_path=os.path.join(cache_location, 'legit_00003'))

        nose.tools.assert_true(crash.explorable())  #判断是否可以 explore
        nose.tools.assert_true(crash.one_of(Vulnerability.WRITE_WHAT_WHERE)) # 漏洞是否为任意写

        crash.explore() #进行探索m
        arsenal = crash.exploit(blacklist_techniques={'rop_set_register', 'rop_leak_memory'})

        nose.tools.assert_true(len(arsenal.register_setters) >= 2)
        nose.tools.assert_true(len(arsenal.leakers) >= 1)

        crash.project.loader.close()

        for reg_setter in arsenal.register_setters:
            nose.tools.assert_true(_do_pov_test(reg_setter))

        for leaker in arsenal.leakers:
            nose.tools.assert_true(_do_pov_test(leaker))

def test_linux_stacksmash_32():
    # Test exploiting a simple linux program with a stack buffer overflow. We should be able to exploit the test binary by
    # ropping to 'system', calling shellcode in the BSS and calling 'jmpsp' shellcode in the BSS.

    inp = b"A" * 227
    lib_path = os.path.join(bin_location, "tests/i386")
    ld_path = os.path.join(lib_path, "ld-linux.so.2")
    path = os.path.join(lib_path, "vuln_stacksmash")
    with archr.targets.LocalTarget([ld_path, '--library-path', lib_path, path], path, target_arch='i386').build().start() as target:
        crash = rex.Crash(target, inp, fast_mode=True, rop_cache_path=os.path.join(cache_location, 'vuln_stacksmash'))

        exploit = crash.exploit(blacklist_techniques={'rop_leak_memory', 'rop_set_register'})
        crash.project.loader.close()

        # make sure we're able to exploit it in all possible ways
        assert len(exploit.arsenal) == 3
        assert 'rop_to_system' in exploit.arsenal
        assert 'call_shellcode' in exploit.arsenal
        assert 'call_jmp_sp_shellcode' in exploit.arsenal

        _check_arsenal_has_send(exploit.arsenal)