1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
import logging
from miasm2.ir.symbexec import SymbolicExecutionEngine
from miasm2.expression.expression import ExprMem
from miasm2.expression.expression_helper import possible_values
from miasm2.expression.simplifications import expr_simp
from miasm2.ir.ir import IRBlock, AssignBlock
LOG_CST_PROPAG = logging.getLogger("cst_propag")
CONSOLE_HANDLER = logging.StreamHandler()
CONSOLE_HANDLER.setFormatter(logging.Formatter("%(levelname)-5s: %(message)s"))
LOG_CST_PROPAG.addHandler(CONSOLE_HANDLER)
LOG_CST_PROPAG.setLevel(logging.WARNING)
class SymbExecState(SymbolicExecutionEngine):
"""
State manager for SymbolicExecution
"""
def __init__(self, ir_arch, state):
super(SymbExecState, self).__init__(ir_arch, {})
self.set_state(state)
def add_state(ir_arch, todo, states, addr, state):
"""
Add or merge the computed @state for the block at @addr. Update @todo
@ir_arch: IR instance
@todo: modified block set
@states: dictionnary linking a label to its entering state.
@addr: address of the concidered block
@state: computed state
"""
addr = ir_arch.get_label(addr)
todo.add(addr)
if addr not in states:
states[addr] = state
else:
states[addr] = states[addr].merge(state)
def is_expr_cst(ir_arch, expr):
"""Return true if @expr is only composed of ExprInt and init_regs
@ir_arch: IR instance
@expr: Expression to test"""
elements = expr.get_r(mem_read=True)
for element in elements:
if element.is_mem():
continue
if element.is_id() and element in ir_arch.arch.regs.all_regs_ids_init:
continue
if element.is_int():
continue
return False
else:
# Expr is a constant
return True
class SymbExecStateFix(SymbolicExecutionEngine):
"""
Emul blocks and replace expressions with their corresponding constant if
any.
"""
# Function used to test if an Expression is considered as a constant
is_expr_cst = lambda _, ir_arch, expr: is_expr_cst(ir_arch, expr)
def __init__(self, ir_arch, state, cst_propag_link):
super(SymbExecStateFix, self).__init__(ir_arch, {})
self.set_state(state)
self.cst_propag_link = cst_propag_link
def propag_expr_cst(self, expr):
"""Propagate constant expressions in @expr
@expr: Expression to update"""
elements = expr.get_r(mem_read=True)
to_propag = {}
for element in elements:
# Only ExprId can be safely propagated
if not element.is_id():
continue
value = self.eval_expr(element)
if self.is_expr_cst(self.ir_arch, value):
to_propag[element] = value
return expr_simp(expr.replace_expr(to_propag))
def eval_updt_irblock(self, irb, step=False):
"""
Symbolic execution of the @irb on the current state
@irb: IRBlock instance
@step: display intermediate steps
"""
assignblks = []
for index, assignblk in enumerate(irb):
new_assignblk = {}
links = {}
for dst, src in assignblk.iteritems():
src = self.propag_expr_cst(src)
if dst.is_mem():
ptr = dst.arg
ptr = self.propag_expr_cst(ptr)
dst = ExprMem(ptr, dst.size)
new_assignblk[dst] = src
if assignblk.instr is not None:
for arg in assignblk.instr.args:
new_arg = self.propag_expr_cst(arg)
links[new_arg] = arg
self.cst_propag_link[(irb.label, index)] = links
self.eval_updt_assignblk(assignblk)
assignblks.append(AssignBlock(new_assignblk, assignblk.instr))
self.ir_arch.blocks[irb.label] = IRBlock(irb.label, assignblks)
def compute_cst_propagation_states(ir_arch, init_addr, init_infos):
"""
Propagate "constant expressions" in a function.
The attribute "constant expression" is true if the expression is based on
constants or "init" regs values.
@ir_arch: IntermediateRepresentation instance
@init_addr: analysis start address
@init_infos: dictionnary linking expressions to their values at @init_addr
"""
done = set()
state = SymbExecState.StateEngine(init_infos)
lbl = ir_arch.get_label(init_addr)
todo = set([lbl])
states = {lbl: state}
while todo:
if not todo:
break
lbl = todo.pop()
state = states[lbl]
if (lbl, state) in done:
continue
done.add((lbl, state))
if lbl not in ir_arch.blocks:
continue
symbexec_engine = SymbExecState(ir_arch, state)
addr = symbexec_engine.run_block_at(lbl)
symbexec_engine.del_mem_above_stack(ir_arch.sp)
for dst in possible_values(addr):
value = dst.value
if value.is_mem():
LOG_CST_PROPAG.warning('Bad destination: %s', value)
continue
elif value.is_int():
value = ir_arch.get_label(value)
add_state(ir_arch, todo, states, value,
symbexec_engine.get_state())
return states
def propagate_cst_expr(ir_arch, addr, init_infos):
"""
Propagate "constant expressions" in a @ir_arch.
The attribute "constant expression" is true if the expression is based on
constants or "init" regs values.
@ir_arch: IntermediateRepresentation instance
@addr: analysis start address
@init_infos: dictionnary linking expressions to their values at @init_addr
Returns a mapping between replaced Expression and their new values.
"""
states = compute_cst_propagation_states(ir_arch, addr, init_infos)
cst_propag_link = {}
for lbl, state in states.iteritems():
if lbl not in ir_arch.blocks:
continue
symbexec = SymbExecStateFix(ir_arch, state, cst_propag_link)
symbexec.eval_updt_irblock(ir_arch.blocks[lbl])
return cst_propag_link
|