Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def read(self, address, size):
if arbitrary(self, address):
raise ArbitraryRead(self, address)
as_ = concretise(self, address)
try:
if len(as_) > 1:
e = None
value = bv.Symbol(size, unique_name('read'))
for a in as_:
v = None
for i in range(0, size // 8):
if v is None:
v = self.memory.read_byte(self, a.value + i)
else:
v = self.memory.read_byte(self, a.value + i).concatenate(v)
if e is None:
e = (address == a) & (value == v)
else:
e = e | ((address == a) & (value == v))
self.solver.add(e)
else:
if scoring_function is not None:
n.score = scoring_function(n)
states.append(n)
states.sort(key=lambda x: x.score)
else:
states.append(n)
available_states.release()
except StateException, v:
v.state.log.vulnerability(v)
v.state.log.debug('saving vuln state {}'.format(v.state.id))
serialisation.save('vuln_state_{}'.format(v.state.id), v)
data = ''
if isinstance(v, ArbitraryRead):
v.state.solver.add(v.address == bv.Constant(v.address.size, 0xc01db33f))
s = v.state
m = v.state.solver.model()
for i in range(0, 0x4000):
name = 'ttf_{:x}'.format(i)
if name in m:
data += chr(m[name].value)
else:
data += '#'
print colored(data, 'white', 'on_red', attrs=['bold'])
print data.encode('hex')
with open('font_{}.ttf'.format(v.state.id), 'wb') as tmp:
tmp.write(data)
active_threads.release()
def strcmp(s, cc):
f = cc(s)
str1 = f.params[0]
str2 = f.params[1]
s.log.function_call(f, 'strcmp(str1={}, str2={})', str1, str2)
iter1 = iter(String(s, str1))
iter2 = iter(String(s, str2))
first_smaller = bv.Constant(32, -1)
first_larger = bv.Constant(32, 1)
zero = bv.Constant(32, 0)
characters = []
not_terminated = None
not_already_terminated = bl.Constant(True)
while True:
(char1, constraint1) = next(iter1)
(char2, constraint2) = next(iter2)
not_terminated = not_already_terminated & constraint1
not_terminated = not_terminated & constraint2
not_terminated = not_terminated & (char1 == char2)
characters.append((not_already_terminated, char1, char2))
not_already_terminated = not_terminated
def concretise(state, value, count=8):
values = set()
constraint = None
if not value.symbolic:
return [value]
elif not isinstance(value, bv.Symbol):
new_value = bv.Symbol(value.size, unique_name('concretise'))
constraint = (new_value == value)
state.solver.add(constraint)
value = new_value
# we now know that value is a symbol
# TODO: this really hurts performance, but it will probably also help
# with finding bugs... add in again once I have better path culling
# heuristics again
#values.add(maximum(state, value))
#values.add(minimum(state, value))
#if len(values) == 1:
# max == min, our work here is done...
def op_bisnz(i, s):
a = operand_value(s, i.input0)
dst = i.output
result = bv.if_then_else(
a != bv.Constant(a.size, 0),
bv.Constant(dst.size, 1),
bv.Constant(dst.size, 0))
s.registers[dst.name] = result
return [s]
def operand_value(s, o):
output = o
if isinstance(o, reil.ImmediateOperand):
output = bv.Constant(o.size, o.value)
elif isinstance(o, reil.RegisterOperand):
output = s.registers[o.name]
if output.size > o.size:
output = output.extract(o.size)
elif output.size < o.size:
output = output.zero_extend_to(o.size)
return output
def read_byte(self, state, address):
try:
return self._cache[address]
except KeyError:
for b, l, d in self._pages:
if b <= address < l:
value = bv.Constant(8, ord(d[address - b]))
self._cache[address] = value
return value
state.throw(InvalidRead(state, address))
if stream.value > len(s.files):
return f.ret(value=0)
else:
file = s.files[stream.value]
offset = file['offset']
output = OutputBuffer(s, buf)
fd = None
if file['path'] not in ['stdin', 'stdout', 'stderr']:
fd = open(file['path'], 'rb')
if size.symbolic or count.symbolic:
raise NotImplementedError()
elif fd is None:
for i in xrange(0, size.value * count.value):
output.append(bv.Symbol(8, 'file_{}_{:x}'.format(stream.value, offset)))
offset += 1
else:
fd.seek(offset, 0)
for i in range(0, size.value * count.value):
byte = fd.read(1)
if len(byte) == 1:
if byte == '#':
output.append(bv.Symbol(8, 'file_{}_{:x}'.format(stream.value, offset)))
else:
output.append(bv.Constant(8, ord(byte)))
offset += 1
else:
break
file['offset'] = offset
if fd.value > len(s.files):
return f.ret(value=0)
else:
file = s.files[fd.value]
offset = file['offset']
output = OutputBuffer(s, buf)
real_fd = None
if file['path'] not in ['stdin', 'stdout', 'stderr']:
real_fd = open(file['path'], 'rb')
if size.symbolic:
raise NotImplementedError()
elif real_fd is None:
for i in xrange(0, size.value):
b = bv.Symbol(8, 'file_{}_{:x}'.format(fd.value, offset))
output.append(b)
file['bytes'][offset] = b
offset += 1
else:
real_fd.seek(offset, 0)
for i in range(0, size.value):
byte = real_fd.read(1)
if len(byte) == 1:
if byte == '#':
b = bv.Symbol(8, 'file_{}_{:x}'.format(fd.value, offset))
else:
b = bv.Constant(8, ord(byte))
output.append(b)
file['bytes'][offset] = b
offset += 1
else: