DeforaNetworks/manticore
3
0
Fork 0
Code Releases Activity

evm: add per state trace file (#817)

Browse Source 
* Add trace file

* Make ethersplay compatible trace format

* Small cleanup

* Record separate init and rt traces, emit separate trace files

* add todo

* More pythonic

* Add test for end insn in trace file

* Fix test

* Move trace accumulation into will_evm_execute callback

here, we can easily and correctly check the .last_exception field
to see if we were in init code or rt code. from the did_evm_execute hook
the last instruction (end instruction) had a different .last_exception
so this was causing the last instruction to not be recorded in the trace.
This commit is contained in:
Mark Mossberg
2018-03-19 14:22:08 -04:00
committed by GitHub
parent 8acd2293b8
commit 2eb002593e
2 changed files with 81 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
manticore/ethereum.py
+33 -17
View File
@@ -138,9 +138,8 @@ class UninitializedStorage(Detector):
state.context.setdefault('seth.detectors.initialized_storage', set()).add(offset)
def calculate_coverage(code, seen):
''' Calculates what percentage of code has been seen '''
runtime_bytecode = code
def calculate_coverage(runtime_bytecode, seen):
''' Calculates what percentage of runtime_bytecode has been seen '''
end = None
if ''.join(runtime_bytecode[-44: -34]) == '\x00\xa1\x65\x62\x7a\x7a\x72\x30\x58\x20' \
and ''.join(runtime_bytecode[-2:]) == '\x00\x29':
@@ -674,7 +673,6 @@ class ManticoreEVM(Manticore):
self._executor.subscribe('did_load_state', self._load_state_callback)
self._executor.subscribe('will_terminate_state', self._terminate_state_callback)
self._executor.subscribe('will_execute_instruction', self._will_execute_instruction_callback)
self._executor.subscribe('did_execute_instruction', self._did_execute_instruction_callback)
self._executor.subscribe('did_read_code', self._did_evm_read_code)
self._executor.subscribe('on_symbolic_sha3', self._symbolic_sha3)
self._executor.subscribe('on_concrete_sha3', self._concrete_sha3)
@@ -1106,17 +1104,16 @@ class ManticoreEVM(Manticore):
assert state.platform.constraints == state.platform.current.constraints
logger.debug("%s", state.platform.current)
if 'Call' in str(type(state.platform.current.last_exception)):
coverage_context_name = 'runtime_coverage'
else:
if isinstance(state.platform.current.last_exception, evm.Create):
coverage_context_name = 'init_coverage'
trace_context_name = 'seth.init.trace'
else:
coverage_context_name = 'runtime_coverage'
trace_context_name = 'seth.rt.trace'
with self.locked_context(coverage_context_name, set) as coverage:
coverage.add((state.platform.current.address, state.platform.current.pc))
def _did_execute_instruction_callback(self, state, prev_pc, pc, instruction):
''' INTERNAL USE '''
state.context.setdefault('seth.trace', []).append((state.platform.current.address, prev_pc))
state.context.setdefault(trace_context_name, []).append((state.platform.current.address, pc))
def _did_evm_read_code(self, state, offset, size):
''' INTERNAL USE '''
@@ -1168,7 +1165,7 @@ class ManticoreEVM(Manticore):
with testcase.open_stream('summary') as summary:
summary.write("Last exception: %s\n" % state.context['last_exception'])
address, offset = state.context['seth.trace'][-1]
address, offset = state.context['seth.rt.trace'][-1]
# Last instruction
metadata = self.get_metadata(blockchain.transactions[-1].address)
@@ -1198,14 +1195,14 @@ class ManticoreEVM(Manticore):
summary.write("\t%032x -> %032x %s\n" % (offset, value, flagged(is_storage_symbolic)))
is_something_symbolic = is_something_symbolic or is_storage_symbolic
code = blockchain.get_code(account_address)
if len(code):
runtime_code = blockchain.get_code(account_address)
if runtime_code:
summary.write("Code:\n")
fcode = StringIO.StringIO(code)
fcode = StringIO.StringIO(runtime_code)
for chunk in iter(lambda: fcode.read(32), b''):
summary.write('\t%s\n' % chunk.encode('hex'))
trace = set((offset for address_i, offset in state.context['seth.trace'] if address == address_i))
summary.write("Coverage %d%% (on this state)\n" % calculate_coverage(code, trace)) # coverage % for address in this account/state
runtime_trace = set((pc for contract, pc in state.context['seth.rt.trace'] if address == contract))
summary.write("Coverage %d%% (on this state)\n" % calculate_coverage(runtime_code, runtime_trace)) # coverage % for address in this account/state
summary.write("\n")
if blockchain._sha3:
@@ -1295,6 +1292,25 @@ class ManticoreEVM(Manticore):
logger.debug("Using iterpickle to dump state")
statef.write(iterpickle.dumps(state, 2))
with testcase.open_stream('rt.trace') as f:
self._emit_trace_file(f, state.context['seth.rt.trace'])
with testcase.open_stream('init.trace') as f:
self._emit_trace_file(f, state.context['seth.init.trace'])
@staticmethod
def _emit_trace_file(filestream, trace):
"""
:param filestream: file object for the workspace trace file
:param trace: list of (contract address, pc) tuples
:type trace: list[tuple(int, int)]
"""
for contract, pc in trace:
if pc == 0:
filestream.write('---\n')
ln = '0x{:x}:0x{:x}\n'.format(contract, pc)
filestream.write(ln)
def finalize(self):
"""
Terminate and generate testcases for all currently alive states (contract states that cleanly executed
tests/test_eth.py
+48 -1
View File
@@ -1,11 +1,13 @@
import unittest
import os
from manticore.core.plugin import Plugin
from manticore.core.smtlib import ConstraintSet, operators
from manticore.core.smtlib.expression import BitVec
from manticore.core.state import State
from manticore.ethereum import ManticoreEVM, IntegerOverflow, Detector, NoAliveStates
from manticore.platforms.evm import EVMWorld, ConcretizeStack, concretized_args
from manticore.platforms.evm import EVMWorld, ConcretizeStack, Create, concretized_args
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -92,6 +94,51 @@ class EthTests(unittest.TestCase):
self.assertIn('STOP', context)
self.assertIn('REVERT', context)
def test_end_instruction_trace(self):
"""
Make sure that the trace files are correct, and include the end instructions
"""
class TestPlugin(Plugin):
"""
Record the pcs of all end instructions encountered. Source of truth.
"""
def will_evm_execute_instruction_callback(self, state, instruction, arguments):
if isinstance(state.platform.current.last_exception, Create):
name = 'init'
else:
name = 'rt'
# collect all end instructions based on whether they are in init or rt
if instruction.semantics in ('REVERT', 'STOP', 'RETURN'):
with self.locked_context(name) as d:
d.append(state.platform.current.pc)
mevm = ManticoreEVM()
p = TestPlugin()
mevm.register_plugin(p)
filename = os.path.join(THIS_DIR, 'binaries/int_overflow.sol')
mevm.multi_tx_analysis(filename, tx_limit=1)
worksp = mevm.workspace
listdir = os.listdir(worksp)
def get_concatenated_files(directory, suffix):
paths = [os.path.join(directory, f) for f in listdir if f.endswith(suffix)]
concatenated = ''.join(open(path).read() for path in paths)
return concatenated
all_init_traces = get_concatenated_files(worksp, 'init.trace')
all_rt_traces = get_concatenated_files(worksp, 'rt.trace')
# make sure all init end insns appear somewhere in the init traces
for pc in p.context['init']:
self.assertIn(':0x{:x}'.format(pc), all_init_traces)
# and all rt end insns appear somewhere in the rt traces
for pc in p.context['rt']:
self.assertIn(':0x{:x}'.format(pc), all_rt_traces)
def test_graceful_handle_no_alive_states(self):
"""
If there are no alive states, or no initial states, we should not crash. issue #795