23c2585316
* WIP New Policy class * WIP pubsub * Update Signal tests * small fixes from github comments * Fix event decode_instruction signature * Good merge * Good good merge * WIP manticore refactor * Fix default old-style initial state * add -> enqueue * @m.init * Fix workspace url * Some test skipped * Ad Fixme to platform specific stuff in State * add -> enqueue * Enqueue created state * Fix m.init Use a messy hack to adhere to the spec (callback func receive 1 state argument) * Add _coverage_file ivar to Manticore * Fix symbolic files * remove extra enqueue * Fixing __main__ * comments * Refactor CLI, and Manticore high level interfaces (#498) * Refactor main, - classmethod for linux - refactor manticore ctor - compat with old linux behavior - changed verbosity API (to allow for this use case: what if you want to set verbosity for the stuff manticore does in its ctor?) * rm old verbosity * small * Add decree classmethod * Rm checks ; they are redundant anyway * Misc * Move add_symbolic_file to linux platform * rm redundant checks * Rm explicit args for deprecated interface * Fix cli bug * Allow for both linux and decree from cli * Add back argv positional param for deprecated api compat
115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
import unittest
|
|
import sys
|
|
import shutil
|
|
import tempfile
|
|
import os
|
|
import hashlib
|
|
import subprocess
|
|
import json
|
|
import time
|
|
|
|
#logging.basicConfig(filename = "test.log",
|
|
# format = "%(asctime)s: %(name)s:%(levelname)s: %(message)s",
|
|
# level = logging.DEBUG)
|
|
|
|
class IntegrationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
# Create a temporary directory
|
|
self.test_dir = tempfile.mkdtemp()
|
|
|
|
def tearDown(self):
|
|
# Remove the directory after the test
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
def _getDumpParams(self, jsonf):
|
|
self.assertTrue(os.path.exists(jsonf))
|
|
|
|
c = open(jsonf, 'r').read()
|
|
return json.loads(c)
|
|
|
|
def _loadVisitedSet(self, visited):
|
|
|
|
self.assertTrue(os.path.exists(visited))
|
|
vitems = open(visited, 'r').read().splitlines()
|
|
|
|
vitems = map(lambda x: int(x[2:], 16), vitems)
|
|
|
|
return set(vitems)
|
|
|
|
def _runWithTimeout(self, procargs, timeout=600):
|
|
|
|
with open(os.path.join(os.pardir, "logfile"), "w") as output:
|
|
#with open(os.path.join(os.pardir, "/dev/stdout"), "w") as output:
|
|
po = subprocess.Popen(procargs, stdout=output)
|
|
secs_used = 0
|
|
|
|
while po.poll() is None and secs_used < timeout:
|
|
time.sleep(1)
|
|
sys.stderr.write("~")
|
|
secs_used += 1
|
|
|
|
self.assertTrue(secs_used < timeout)
|
|
sys.stderr.write("\n")
|
|
|
|
def _runManticore(self, dumpname):
|
|
|
|
dirname = os.path.dirname(__file__)
|
|
dumpdir = os.path.abspath(os.path.join(dirname, 'memdumps', dumpname))
|
|
|
|
self.assertTrue(os.path.exists(dumpdir))
|
|
|
|
jsonfile = os.path.join(dumpdir, 'args.json')
|
|
|
|
params = self._getDumpParams(jsonfile)
|
|
|
|
workspace = os.path.join(self.test_dir, 'ws_{}'.format(dumpname))
|
|
logfile = os.path.join(workspace, "output.log")
|
|
|
|
dumpfile = os.path.join(dumpdir, params['dump'])
|
|
|
|
args = ['python', '-m', 'manticore', '--timeout', '400', '--workspace', workspace, dumpfile]
|
|
|
|
os.mkdir(workspace)
|
|
|
|
for k,v in params.iteritems():
|
|
if k.startswith("--"):
|
|
args.extend([k, v.format(dumpdir=dumpdir, workspace=workspace)])
|
|
self._runWithTimeout(args, logfile)
|
|
|
|
efile = os.path.join(dumpdir, params['expected'])
|
|
expected = self._loadVisitedSet(efile)
|
|
|
|
afile = os.path.join(workspace, params['actual'])
|
|
actual = self._loadVisitedSet(afile)
|
|
|
|
self.assertGreaterEqual(actual, expected)
|
|
|
|
@unittest.skip('Windows temporarily unmaintained')
|
|
def testSimpleParse(self):
|
|
self._runManticore("simple_parse")
|
|
|
|
@unittest.skip('Windows temporarily unmaintained')
|
|
def testSimpleDeref(self):
|
|
self._runManticore("simple_bad_deref")
|
|
|
|
@unittest.skip('Windows temporarily unmaintained')
|
|
def testSimpleBufferOverflow(self):
|
|
self._runManticore("simple_buffer_overflow")
|
|
|
|
# generate too many states on memory concretization
|
|
#def testSimpleFpu(self):
|
|
# self._runManticore("simple_fpu")
|
|
|
|
# too slow processing REP SCASD
|
|
@unittest.skip('TODO')
|
|
def testWin32API(self):
|
|
self._runManticore("win32_api_test")
|
|
|
|
@unittest.skip('Windows temporarily unmaintained')
|
|
def testAPIInterception(self):
|
|
self._runManticore("api_interception")
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|