Angr 中的函数识别
函数识别概述
二进制文件通过IDA或者radare2这样的反汇编工具，能够识别二进制文件中的函数边界信息，并根据其调用关系生成整个程序的函数调用图。因此准确识别二进制文件中的函数边界对于进一步的分析二进制文件非常重要。二进制文件通常组织为 数据、代码、元数据的形式，在没有strip掉符号表的二进制文件中，关于函数的起始偏移、大小通常在元数据中可以直接找到，但是对于去除符号表的二进制文件，由于其符号表的缺失，通常需要采用别的以下方法进行函数识别：
-
函数序言和尾声
-
通过对call、jmp等指令所跳转的地址进行分析
-
分析不可达代码位置
但是采用以上方法同样会存在一些不准确的问题，比如依赖函数序言和尾声的检查，在使用不同编译器或者优化选项的情况下，采用这种硬编码的方式可能就不再适用；对于call或者jmp指令的分析，可能很多地址只有在运行时才能确定等等。
Angr中的函数识别
1. 初始化阶段
Angr中函数识别的过程是在CFGFast中构建CFG的时候进行的，在初始化阶段主要步骤如下：
-
首先调用
self._executable_memory_regions获取二进制文件有执行权限的region，将这些区间作为候选分析位置 -
然后对得到的regions进行一些筛选和处理，同时对分析进行一些设置，比如设置是否有符号表、是否收集数据引用信息等。这里可以使用objdump命令查看文件的符号表信息:
- 查看动态符号表信息
nevv@ubuntu:~/angr/others$ objdump -T cgibin | grep "main" 004023e0 g DF .text 00000304 Base main 00409480 g DF .text 000005ec Base hedwigcgi_main 00408a78 g DF .text 00000198 Base captchacgi_main 0040b930 g DF .text 00000b28 Base conntrackcgi_main 00406b40 g DF .text 00000008 Base dlcfg_main 0040a268 g DF .text 000003d4 Base servicecgi_main 0040cb5c g DF .text 00000604 Base hnap_main 0040a6f0 g DF .text 00000218 Base ssdpcgi_main 0040b818 g DF .text 00000110 Base genacgi_main 00407164 g DF .text 000000e4 Base fwupdater_main 00409ae0 g DF .text 00000620 Base pigwidgeoncgi_main 00406b48 g DF .text 00000008 Base fwup_main 004195a0 DF *UND* 00000000 __uClibc_main 00405284 g DF .text 000002c8 Base phpcgi_main 0040ae50 g DF .text 00000484 Base soapcgi_main 00408e10 g DF .text 0000066c Base sessioncgi_main 00406b50 g DF .text 00000008 Base seamacgi_main- 查看静态符号表信息:
nevv@ubuntu:~/angr/others$ objdump -t cgibin cgibin: file format elf32-tradlittlemips SYMBOL TABLE: no symbols两者的区别如下：
-
静态链接中有一个专门的段叫符号表 -- “.symtab”(Symbol Table)， 里面保存了所有关于该目标文件的符号的定义和引用。
-
动态链接中同样有一个段叫 动态符号表 -- “.dynsym”(Dynamic Symbol) ， 但.dynsym 相对于 .symtab 只保存了与动态链接相关的导入导出符号。so中同样有.symtab，其中保存着所有的符号。
-
最后调用
self._analyze进行前向分析
2. 分析阶段
_analyze方法很简单，主要就是self._pre_analysis()和根据self._graph_visitor 是否存在来调用对应的分析方法，对于还没有建立一个图结构的分析来说（就比如cfg恢复分析），第一次总是会调用self._analysis_core_baremetal() 方法。
def _analyze(self):
"""
The main analysis routine.
:return: None
"""
self._pre_analysis()
if self._graph_visitor is None:
# There is no base graph that we can rely on. The analysis itself should generate successors for the
# current job.
# An example is the CFG recovery.
self._analysis_core_baremetal()
else:
# We have a base graph to follow. Just handle the current job.
self._analysis_core_graph()
self._post_analysis()
0x1 _pre_analysis
这个函数的主要功能就是，初始化分析过程中需要用到的变量，使用符号表，将符号表中的函数起始位置作为分析的起始位置，使用函数序言搜索函数，并将搜索结果保存在 _function_prologue_addrs 中。
def _pre_analysis(self):
import pdb
pdb.set_trace()
# 初始化一些cfg相关的变量
self._initialize_cfg()
# Scan for __x86_return_thunk and friends
self._known_thunks = self._find_thunks()
"""
这里应该是会寻找一些特殊的字节序列
>>> print disasm('E807000000F3900FAEE8EBF948890424C3'.decode("hex"))
0: e8 07 00 00 00 call 0xc
5: f3 90 pause
7: 0f ae e8 lfence
a: eb f9 jmp 0x5
c: 48 dec eax
d: 89 04 24 mov DWORD PTR [esp],eax
10: c3 ret
>>> print disasm('E807000000F3900FAEE8EBF9488D642408C3'.decode("hex"))
0: e8 07 00 00 00 call 0xc
5: f3 90 pause
7: 0f ae e8 lfence
a: eb f9 jmp 0x5
c: 48 dec eax
d: 8d 64 24 08 lea esp,[esp+0x8]
11: c3 ret
>>>
"""
# 初始化一些分析时候需要用到的变量
self._pending_jobs = PendingJobs(self.functions, self._deregister_analysis_job)
self._traced_addresses = set()
self._function_returns = defaultdict(set)
# 不是所有的函数调用都使用call指令，因此需要记录下每一个单一函数的退出点，
# 在需要函数调用的时候，在函数调用图上添加对应的边
self._function_exits = defaultdict(set)
# 创建一个初始化状态
self._initial_state = self.project.factory.blank_state(mode="fastpath")
initial_options = self._initial_state.options - {o.TRACK_CONSTRAINTS} - o.refs
initial_options |= {o.SUPER_FASTPATH, o.SYMBOL_FILL_UNCONSTRAINED_REGISTERS, o.SYMBOL_FILL_UNCONSTRAINED_MEMORY}
# initial_options.remove(o.COW_STATES)
self._initial_state.options = initial_options
starting_points = set()
# clear all existing functions
self.kb.functions.clear()
if self._use_symbols:
starting_points |= self._function_addresses_from_symbols
# 根据符号表获取函数的地址，这里有190个函数
"""
调试的时候不知道这里为什么有一个0
"""
if self._extra_function_starts:
starting_points |= set(self._extra_function_starts)
# 对函数入口点进行排序
starting_points = sorted(list(starting_points), reverse=True)
if self._start_at_entry and self.project.entry is not None and self._inside_regions(self.project.entry) and \
self.project.entry not in starting_points:
# make sure self.project.entry is inserted
starting_points += [ self.project.entry ]
# 对于每一个起始位置，创建一个CFGJOB进行分析
for sp in starting_points:
job = CFGJob(sp, sp, 'Ijk_Boring')
self._insert_job(job)
# register the job to function `sp`
self._register_analysis_job(sp, job)
self._updated_nonreturning_functions = set()
# 这里是使用函数序言进行函数查找，该例子找到了217个
if self._use_function_prologues and self.project.concrete_target is None:
self._function_prologue_addrs = sorted(self._func_addrs_from_prologues())
# make a copy of those prologue addresses, so that we can pop from the list
self._remaining_function_prologue_addrs = self._function_prologue_addrs[::]
# make function_prologue_addrs a set for faster lookups
self._function_prologue_addrs = set(self._function_prologue_addrs)
0x2 _analysis_core_baremetal
这个函数的功能就是从刚才的队列中取出来添加的 job 并处理。主要就是以下三步：
-
_pre_job_handling
-
_process_job_and_get_successors
-
_intra_analysis
该函数的源码如下：
def _analysis_core_baremetal(self):
if not self._job_info_queue:
self._job_queue_empty()
while not self.should_abort:
if self._status_callback is not None:
self._status_callback(self)
# should_abort might be changed by the status callback function
if self.should_abort:
return
if not self._job_info_queue:
self._job_queue_empty() # 时间消耗 1/3
if not self._job_info_queue:
# still no job available
break
job_info = self._job_info_queue[0]
try:
self._pre_job_handling(job_info.job)
except AngrDelayJobNotice:
# delay the handling of this job
continue
except AngrSkipJobNotice:
# consume and skip this job
self._job_info_queue = self._job_info_queue[1:]
self._job_map.pop(self._job_key(job_info.job), None)
continue
# remove the job info from the map
self._job_map.pop(self._job_key(job_info.job), None)
self._job_info_queue = self._job_info_queue[1:]
self._process_job_and_get_successors(job_info)
# Short-cut for aborting the analysis
if self.should_abort:
break
self._intra_analysis()
- _job_queue_empty 占了构建cfg的总时间的1/3。其执行流程如下
- 首先查找来自必须要返回的函数的job，如果有，添加到队列中
- 然后将已经完成分析的函数添加到_completed_functions中
- 迭代地分析所有更改的函数，更新它们的返回属性，直到达到一个定点(即没有找到新的返回/不返回函数)。
- 尝试分析剩下的间接调用位置
- 如果选择使用函数序言分析的话，就将之前使用函数序言找到的函数添加到job队列中
- 如果选择使用完整分析的话，会获取下条指令封装为job添加到队列中
def _job_queue_empty(self):
if self._pending_jobs:
# fastpath
# look for a job that comes from a function that must return
# if we can find one, just use it
job = self._pop_pending_job(returning=True)
if job is not None:
self._insert_job(job)
return
self._clean_pending_exits()
# did we finish analyzing any function?
# fill in self._completed_functions
self._make_completed_functions()
# analyze function features, most importantly, whether each function returns or not
self._analyze_all_function_features()
# Clear _changed_functions set
self._updated_nonreturning_functions = set()
if self._pending_jobs:
self._clean_pending_exits()
job = self._pop_pending_job(returning=True)
if job is not None:
self._insert_job(job)
return
job = self._pop_pending_job(returning=False)
if job is not None:
self._insert_job(job)
return
# Try to see if there is any indirect jump left to be resolved
if self._resolve_indirect_jumps and self._indirect_jumps_to_resolve:
self._process_unresolved_indirect_jumps()
if self._job_info_queue:
return
if self._use_function_prologues and self._remaining_function_prologue_addrs:
while self._remaining_function_prologue_addrs:
prolog_addr = self._remaining_function_prologue_addrs[0]
self._remaining_function_prologue_addrs = self._remaining_function_prologue_addrs[1:]
if self._seg_list.is_occupied(prolog_addr):
continue
job = CFGJob(prolog_addr, prolog_addr, 'Ijk_Boring')
self._insert_job(job)
self._register_analysis_job(prolog_addr, job)
return
if self._force_complete_scan:
addr = self._next_code_addr()
if addr is None:
l.debug("Force-scan jumping failed")
else:
l.debug("Force-scanning to %#x", addr)
if addr is not None:
job = CFGJob(addr, addr, "Ijk_Boring", last_addr=None, job_type=CFGJob.JOB_TYPE_COMPLETE_SCANNING)
self._insert_job(job)
self._register_analysis_job(addr, job)
-
_pre_job_handling
一些简单的对 job 的预处理，比如对进度条的计算
-
_process_job_and_get_successors
这个函数的主要目的是对 job 进行处理，并获取当前job的后继，把当前job的后继节点包装为 job，添加入队列。时间消耗约为2/3
def _process_job_and_get_successors(self, job_info):
"""
Process a job, get all successors of this job, and call _handle_successor() to handle each successor.
:param JobInfo job_info: The JobInfo instance
:return: None
"""
job = job_info.job
successors = self._get_successors(job)
all_new_jobs = [ ]
for successor in successors:
new_jobs = self._handle_successor(job, successor, successors)
# 在cfgfast中是直接把其所有的后继返回
if new_jobs:
all_new_jobs.extend(new_jobs)
for new_job in new_jobs:
self._insert_job(new_job)
self._post_job_handling(job, all_new_jobs, successors)
_get_successors 在子类 cfgFast 中实现，主要功能是从给定的地址在搜索一个基本块
0x3 _get_successors
调用_scan_block函数，并将其后继基本块包装为job对象，并添加到待分析的队列中:
def _get_successors(self, job): # pylint:disable=arguments-differ
# current_function_addr = job.func_addr
# addr = job.addr
# if current_function_addr != -1:
# l.debug("Tracing new exit %#x in function %#x", addr, current_function_addr)
# else:
# l.debug("Tracing new exit %#x", addr)
jobs = self._scan_block(job)
# l.debug("... got %d jobs: %s", len(jobs), jobs)
for job_ in jobs: # type: CFGJob
# register those jobs
self._register_analysis_job(job_.func_addr, job_)
return jobs
0x4 _scan_block
def _scan_block(self, cfg_job):
"""
Scan a basic block starting at a specific address
:param CFGJob cfg_job: The CFGJob instance.
:return: a list of successors
:rtype: list
"""
addr = cfg_job.addr
current_func_addr = cfg_job.func_addr
# Fix the function address
# This is for rare cases where we cannot successfully determine the end boundary of a previous function, and
# as a consequence, our analysis mistakenly thinks the previous function goes all the way across the boundary,
# resulting the missing of the second function in function manager.
if addr in self._function_addresses_from_symbols:
current_func_addr = addr
if self._addr_hooked_or_syscall(addr):
entries = self._scan_procedure(cfg_job, current_func_addr)
else:
entries = self._scan_irsb(cfg_job, current_func_addr)
return entries
0x5 _scan_irsb
def _scan_irsb(self, cfg_job, current_func_addr):
"""
Generate a list of successors (generating them each as entries) to IRSB.
Updates previous CFG nodes with edges.
:param CFGJob cfg_job: The CFGJob instance.
:param int current_func_addr: Address of the current function
:return: a list of successors
:rtype: list
"""
# 生成cfgnode
addr, function_addr, cfg_node, irsb = self._generate_cfgnode(cfg_job, current_func_addr)
# 添加函数内部指向该node的边
cfg_job.apply_function_edges(self, clear=True)
# function_addr and current_function_addr can be different. e.g. when tracing an optimized tail-call that jumps
# into another function that has been identified before.
if cfg_node is None:
# exceptions occurred, or we cannot get a CFGNode for other reasons
return [ ]
# 为cfg添加相应的边
self._graph_add_edge(cfg_node, cfg_job.src_node, cfg_job.jumpkind, cfg_job.src_ins_addr,
cfg_job.src_stmt_idx
)
# 将对应的cfg添加到对应函数
self._function_add_node(cfg_node, function_addr)
if self.functions.get_by_addr(function_addr).returning is not True:
self._updated_nonreturning_functions.add(function_addr)
# If we have traced it before, don't trace it anymore
real_addr = get_real_address_if_arm(self.project.arch, addr)
if real_addr in self._traced_addresses:
# the address has been traced before
return [ ]
else:
# Mark the address as traced
self._traced_addresses.add(real_addr)
# irsb cannot be None here
# assert irsb is not None
# IRSB在每个CFGNode中只使用一次，因此在这里必须释放掉以节省内存
cfg_node.irsb = None
# 1/10 _scan_irsb 的时间消耗
self._process_block_arch_specific(addr, irsb, function_addr)
# Scan the basic block to collect data references
if self._collect_data_ref:
self._collect_data_references(irsb, addr)
# 3/20 _scan_irsb 的时间消耗
# Get all possible successors
irsb_next, jumpkind = irsb.next, irsb.jumpkind
successors = [ ]
last_ins_addr = None
ins_addr = addr
if irsb.statements:
for i, stmt in enumerate(irsb.statements):
if isinstance(stmt, pyvex.IRStmt.Exit):
successors.append((i,
last_ins_addr if self.project.arch.branch_delay_slot else ins_addr,
stmt.dst,
stmt.jumpkind
)
)
elif isinstance(stmt, pyvex.IRStmt.IMark):
last_ins_addr = ins_addr
ins_addr = stmt.addr + stmt.delta
else:
for ins_addr, stmt_idx, exit_stmt in irsb.exit_statements:
successors.append((
stmt_idx,
last_ins_addr if self.project.arch.branch_delay_slot else ins_addr,
exit_stmt.dst,
exit_stmt.jumpkind
))
successors.append((DEFAULT_STATEMENT,
last_ins_addr if self.project.arch.branch_delay_slot else ins_addr, irsb_next, jumpkind)
)
entries = [ ]
# 如果是arm架构的话，就会做一些处理，然后再返回
successors = self._post_process_successors(addr, irsb.size, successors)
# Process each successor 这一部分时间消耗占用了 15/20
for suc in successors:
stmt_idx, ins_addr, target, jumpkind = suc
entries += self._create_jobs(target, jumpkind, function_addr, irsb, addr, cfg_node, ins_addr,
stmt_idx
)
return entries
0x6 _create_jobs
给定一个node和其后继节点的一些信息，返回CFGJobs的list。这个函数的主要执行流程如下：
-
根据当前cfg_node获取target_address
-
如果target_address是None，说明跳转地址不是一个确定的值，此时根据jumpkind的类型进行判断：
-
如果是ret类型，说明是函数结尾，否则说明是一个间接跳转，调用_indirect_jump_encountered尝试去解析这个间接跳转
-
如果成功解析，则直接把其所有可能的地址包装为cfg_job并返回
-
如果jumpkind 属于 ("Ijk_Boring", 'Ijk_InvalICache')，说明跳转的地址是一个plt表的地址，直接添加到对应plt函数的边，并创建一个CFGJob添加到jobs中
-
否则就把这个间接跳转加入待解析的间接跳转工作队列中。
-
def _create_jobs(self, target, jumpkind, current_function_addr, irsb, addr, cfg_node, ins_addr, stmt_idx):
"""
Given a node and details of a successor, makes a list of CFGJobs
and if it is a call or exit marks it appropriately so in the CFG
:param int target: Destination of the resultant job
:param str jumpkind: The jumpkind of the edge going to this node
:param int current_function_addr: Address of the current function
:param pyvex.IRSB irsb: IRSB of the predecessor node
:param int addr: The predecessor address
:param CFGNode cfg_node: The CFGNode of the predecessor node
:param int ins_addr: Address of the source instruction.
:param int stmt_idx: ID of the source statement.
:return: a list of CFGJobs
:rtype: list
"""
if type(target) is pyvex.IRExpr.Const: # pylint: disable=unidiomatic-typecheck
target_addr = target.con.value
elif type(target) in (pyvex.IRConst.U8, pyvex.IRConst.U16, pyvex.IRConst.U32, pyvex.IRConst.U64): # pylint: disable=unidiomatic-typecheck
target_addr = target.value
elif type(target) is int: # pylint: disable=unidiomatic-typecheck
target_addr = target
else:
target_addr = None
if target_addr in self._known_thunks and jumpkind == 'Ijk_Boring':
thunk_kind = self._known_thunks[target_addr][0]
if thunk_kind == 'ret':
jumpkind = 'Ijk_Ret'
target_addr = None
elif thunk_kind == 'jmp':
pass # ummmmmm not sure about this one
else:
raise AngrCFGError("This shouldn't be possible")
jobs = [ ]
is_syscall = jumpkind.startswith("Ijk_Sys")
# Special handling:
# If a call instruction has a target that points to the immediate next instruction, we treat it as a boring jump
if jumpkind == "Ijk_Call" and \
not self.project.arch.call_pushes_ret and \
cfg_node.instruction_addrs and \
ins_addr == cfg_node.instruction_addrs[-1] and \
target_addr == irsb.addr + irsb.size:
jumpkind = "Ijk_Boring"
if target_addr is None:
# The target address is not a concrete value
if jumpkind == "Ijk_Ret":
# This block ends with a return instruction.
if current_function_addr != -1:
self._function_exits[current_function_addr].add(addr)
self._function_add_return_site(addr, current_function_addr)
self.functions[current_function_addr].returning = True
self._pending_jobs.add_returning_function(current_function_addr)
cfg_node.has_return = True
elif self._resolve_indirect_jumps and \
(jumpkind in ('Ijk_Boring', 'Ijk_Call', 'Ijk_InvalICache') or jumpkind.startswith('Ijk_Sys')):
# This is an indirect jump. Try to resolve it.
# FIXME: in some cases, a statementless irsb will be missing its instr addresses
# and this next part will fail. Use the real IRSB instead
irsb = cfg_node.block.vex
cfg_node.instruction_addrs = irsb.instruction_addresses
resolved, resolved_targets, ij = self._indirect_jump_encountered(addr, cfg_node, irsb,
current_function_addr, stmt_idx)
if resolved:
for resolved_target in resolved_targets:
if jumpkind == 'Ijk_Call':
jobs += self._create_job_call(cfg_node.addr, irsb, cfg_node, stmt_idx, ins_addr,
current_function_addr, resolved_target, jumpkind)
else:
edge = FunctionTransitionEdge(cfg_node, resolved_target, current_function_addr,
to_outside=False, stmt_idx=stmt_idx, ins_addr=ins_addr,
)
ce = CFGJob(resolved_target, current_function_addr, jumpkind,
last_addr=resolved_target, src_node=cfg_node, src_stmt_idx=stmt_idx,
src_ins_addr=ins_addr, func_edges=[ edge ],
)
jobs.append(ce)
return jobs
if jumpkind in ("Ijk_Boring", 'Ijk_InvalICache'):
resolved_as_plt = False
if irsb and self._heuristic_plt_resolving:
# Test it on the initial state. Does it jump to a valid location?
# It will be resolved only if this is a .plt entry
resolved_as_plt = self._resolve_plt(addr, irsb, ij)
if resolved_as_plt:
jump_target = next(iter(ij.resolved_targets))
target_func_addr = jump_target # TODO: FIX THIS
edge = FunctionTransitionEdge(cfg_node, jump_target, current_function_addr,
to_outside=True, dst_func_addr=jump_target,
stmt_idx=stmt_idx, ins_addr=ins_addr,
)
ce = CFGJob(jump_target, target_func_addr, jumpkind, last_addr=jump_target,
src_node=cfg_node, src_stmt_idx=stmt_idx, src_ins_addr=ins_addr,
func_edges=[edge],
)
jobs.append(ce)
if resolved_as_plt:
# has been resolved as a PLT entry. Remove it from indirect_jumps_to_resolve
if ij.addr in self._indirect_jumps_to_resolve:
self._indirect_jumps_to_resolve.remove(ij.addr)
self._deregister_analysis_job(current_function_addr, ij)
else:
# add it to indirect_jumps_to_resolve
self._indirect_jumps_to_resolve.add(ij)
# register it as a job for the current function
self._register_analysis_job(current_function_addr, ij)
else: # jumpkind == "Ijk_Call" or jumpkind.startswith('Ijk_Sys')
self._indirect_jumps_to_resolve.add(ij)
self._register_analysis_job(current_function_addr, ij)
jobs += self._create_job_call(addr, irsb, cfg_node, stmt_idx, ins_addr, current_function_addr, None,
jumpkind, is_syscall=is_syscall
)
elif target_addr is not None:
# This is a direct jump with a concrete target.
# pylint: disable=too-many-nested-blocks
if jumpkind in ('Ijk_Boring', 'Ijk_InvalICache'):
# if the target address is at another section, it has to be jumping to a new function
if not self._addrs_belong_to_same_section(addr, target_addr):
target_func_addr = target_addr
to_outside = True
else:
# it might be a jumpout
target_func_addr = None
real_target_addr = get_real_address_if_arm(self.project.arch, target_addr)
if real_target_addr in self._traced_addresses:
node = self.model.get_any_node(target_addr)
if node is not None:
target_func_addr = node.function_address
if target_func_addr is None:
target_func_addr = current_function_addr
to_outside = not target_func_addr == current_function_addr
edge = FunctionTransitionEdge(cfg_node, target_addr, current_function_addr,
to_outside=to_outside,
dst_func_addr=target_func_addr,
ins_addr=ins_addr,
stmt_idx=stmt_idx,
)
ce = CFGJob(target_addr, target_func_addr, jumpkind, last_addr=addr, src_node=cfg_node,
src_ins_addr=ins_addr, src_stmt_idx=stmt_idx, func_edges=[ edge ])
jobs.append(ce)
elif jumpkind == 'Ijk_Call' or jumpkind.startswith("Ijk_Sys"):
jobs += self._create_job_call(addr, irsb, cfg_node, stmt_idx, ins_addr, current_function_addr,
target_addr, jumpkind, is_syscall=is_syscall
)
else:
# TODO: Support more jumpkinds
l.debug("Unsupported jumpkind %s", jumpkind)
l.debug("Instruction address: %#x", ins_addr)
return jobs
0x7 时间消耗分析
- 例子： cgibin （D-link DIR 815路由器固件）
- 总时间消耗： 32s （基本上30s全是构建cfg的时间）
- _job_queue_empty 大约占了总时间1/3的时间消耗（10s）
- scan_block 大约占了总时间2/3的时间消耗（20s）
- 其中最耗时的函数是_create_jobs函数，大约占用了16s
- _create_jobs函数主要是处理直接调用和间接调用，间接调用几乎占了_create_jobs函数100%的时间消耗
3. 解析间接调用
0x1 process_unresolved_indirect_jumps
- Ijk_Call 类型表示地址是由前一个基本块传递过来
- Ijk_Boring类型表示是一个jump table类型的
def _process_unresolved_indirect_jumps(self):
"""
Resolve all unresolved indirect jumps found in previous scanning.
Currently we support resolving the following types of indirect jumps:
- Ijk_Call: indirect calls where the function address is passed in from a proceeding basic block
- Ijk_Boring: jump tables
- For an up-to-date list, see analyses/cfg/indirect_jump_resolvers
:return: A set of concrete indirect jump targets (ints).
:rtype: set
"""
l.info("%d indirect jumps to resolve.", len(self._indirect_jumps_to_resolve))
all_targets = set()
for idx, jump in enumerate(self._indirect_jumps_to_resolve): # type:int,IndirectJump
if self._low_priority:
self._release_gil(idx, 20, 0.0001)
all_targets |= self._process_one_indirect_jump(jump)
self._indirect_jumps_to_resolve.clear()
return all_targets
0x2 process_one_indirect_jump
使用angr.analyses.cfg.indirect_jump_resolvers.jumptable.JumpTableResolver求解器来求解间接调用。
def _process_one_indirect_jump(self, jump):
"""
Resolve a given indirect jump.
:param IndirectJump jump: The IndirectJump instance.
:return: A set of resolved indirect jump targets (ints).
"""
resolved = False
resolved_by = None
targets = None
block = self._lift(jump.addr, opt_level=1)
for resolver in self.indirect_jump_resolvers:
resolver.base_state = self._base_state
if not resolver.filter(self, jump.addr, jump.func_addr, block, jump.jumpkind):
continue
resolved, targets = resolver.resolve(self, jump.addr, jump.func_addr, block, jump.jumpkind)
if resolved:
resolved_by = resolver
break
if resolved:
self._indirect_jump_resolved(jump, jump.addr, resolved_by, targets)
else:
self._indirect_jump_unresolved(jump)
return set() if targets is None else set(targets)
0x3 indirect_jump_encountered
当遇到间接跳转时调用。将尝试使用不受时间限制(快速)的间接跳转解析器来解决这个间接跳转。如果无法解决，将查看以前是否已经解决了这个间接跳转。
def _indirect_jump_encountered(self, addr, cfg_node, irsb, func_addr, stmt_idx=DEFAULT_STATEMENT):
"""
Called when we encounter an indirect jump. We will try to resolve this indirect jump using timeless (fast)
indirect jump resolvers. If it cannot be resolved, we will see if this indirect jump has been resolved before.
:param int addr: Address of the block containing the indirect jump.
:param cfg_node: The CFGNode instance of the block that contains the indirect jump.
:param pyvex.IRSB irsb: The IRSB instance of the block that contains the indirect jump.
:param int func_addr: Address of the current function.
:param int or str stmt_idx: ID of the source statement.
:return: A 3-tuple of (whether it is resolved or not, all resolved targets, an IndirectJump object
if there is one or None otherwise)
:rtype: tuple
"""
jumpkind = irsb.jumpkind
l.debug('IRSB %#x has an indirect jump (%s) as its default exit.', addr, jumpkind)
# try resolving it fast
resolved, resolved_targets = self._resolve_indirect_jump_timelessly(addr, irsb, func_addr, jumpkind)
if resolved:
l.debug("Indirect jump at block %#x is resolved by a timeless indirect jump resolver. "
"%d targets found.", addr, len(resolved_targets))
return True, resolved_targets, None
l.debug("Indirect jump at block %#x cannot be resolved by a timeless indirect jump resolver.", addr)
# Add it to our set. Will process it later if user allows.
# Create an IndirectJump instance
if addr not in self.indirect_jumps:
if self.project.arch.branch_delay_slot:
ins_addr = cfg_node.instruction_addrs[-2]
else:
ins_addr = cfg_node.instruction_addrs[-1]
ij = IndirectJump(addr, ins_addr, func_addr, jumpkind, stmt_idx, resolved_targets=[])
self.indirect_jumps[addr] = ij
resolved = False
else:
ij = self.indirect_jumps[addr] # type: IndirectJump
resolved = len(ij.resolved_targets) > 0
return resolved, ij.resolved_targets, ij
0x4 resolve_indirect_jump_timelessly
会调用angr.analyses.cfg.indirect_jump_resolvers.mips_elf_fast.MipsElfFastResolver来求解间接调用。
def _resolve_indirect_jump_timelessly(self, addr, block, func_addr, jumpkind):
"""
Checks if MIPS32 and calls MIPS32 check, otherwise false
:param int addr: irsb address
:param pyvex.IRSB block: irsb
:param int func_addr: Function address
:return: If it was resolved and targets alongside it
:rtype: tuple
"""
if block.statements is None:
block = self.project.factory.block(block.addr, size=block.size).vex
for res in self.timeless_indirect_jump_resolvers:
if res.filter(self, addr, func_addr, block, jumpkind):
r, resolved_targets = res.resolve(self, addr, func_addr, block, jumpkind)
if r:
return True, resolved_targets
return False, [ ]
网友评论