Code cleanup and coverage (#1035)

* dead code elimination, __init__ cleanup
* `binary.Elf` bugfix, add `binary` package tests

Author: defunctio

.coveragerc

@@ -1,3 +1,4 @@
 [run]
-source = .
-omit = tests/*
+source = manticore
+omit =
+	*__init__.py

manticore/binary/__init__.py

@@ -15,124 +15,7 @@
 
 '''
 
-from elftools.elf.elffile import ELFFile
-import io
-
-
-class Binary:
-    magics = {}
-
-    def __new__(cls, path):
-        if cls is Binary:
-            cl = cls.magics[open(path, 'rb').read(4)]
-            return cl(path)
-        else:
-            return super(Binary, cls).__new__(cls)
-
-    def __init__(self, path):
-        self.path = path
-        self.magic = Binary.magics[open(path, 'rb').read(4)]
-
-    def arch(self):
-        pass
-
-    def maps(self):
-        pass
-
-    def threads(self):
-        pass
-
-
-class CGCElf(Binary):
-
-    @staticmethod
-    def _cgc2elf(filename):
-        # hack begin so we can use upstream Elftool
-        with open(filename, 'rb') as fd:
-            stream = io.BytesIO(fd.read())
-            stream.write(b'\x7fELF')
-            stream.name = fd.name
-            return stream
-
-    def __init__(self, filename):
-        super().__init__(filename)
-        stream = self._cgc2elf(filename)
-        self.elf = ELFFile(stream)
-        self.arch = {'x86': 'i386', 'x64': 'amd64'}[self.elf.get_machine_arch()]
-
-        assert 'i386' == self.arch
-        assert self.elf.header.e_type in ['ET_EXEC']
-
-    def maps(self):
-        for elf_segment in self.elf.iter_segments():
-            if elf_segment.header.p_type not in ['PT_LOAD', 'PT_NULL', 'PT_PHDR', 'PT_CGCPOV2']:
-                raise Exception("Not Supported Section")
-
-            if elf_segment.header.p_type != 'PT_LOAD' or elf_segment.header.p_memsz == 0:
-                continue
-
-            flags = elf_segment.header.p_flags
-            # PF_X 0x1 Execute - PF_W 0x2 Write - PF_R 0x4 Read
-            perms = ['   ', '  x', ' w ', ' wx', 'r  ', 'r x', 'rw ', 'rwx'][flags & 7]
-            if 'r' not in perms:
-                raise Exception("Not readable map from cgc elf not supported")
-
-            # CGCMAP--
-            assert elf_segment.header.p_filesz != 0 or elf_segment.header.p_memsz != 0
-            yield((elf_segment.header.p_vaddr,
-                   elf_segment.header.p_memsz,
-                   perms,
-                   elf_segment.stream.name, elf_segment.header.p_offset, elf_segment.header.p_filesz))
-
-    def threads(self):
-        yield(('Running', {'EIP': self.elf.header.e_entry}))
-
-
-class Elf(Binary):
-    def __init__(self, filename):
-        super().__init__(filename)
-        self.elf = ELFFile(open(filename))
-        self.arch = {'x86': 'i386', 'x64': 'amd64'}[self.elf.get_machine_arch()]
-        assert self.elf.header.e_type in ['ET_DYN', 'ET_EXEC', 'ET_CORE']
-
-        # Get interpreter elf
-        self.interpreter = None
-        for elf_segment in self.elf.iter_segments():
-            if elf_segment.header.p_type != 'PT_INTERP':
-                continue
-            self.interpreter = Elf(elf_segment.data()[:-1])
-            break
-        if self.interpreter is not None:
-            assert self.interpreter.arch == self.arch
-            assert self.interpreter.elf.header.e_type in ['ET_DYN', 'ET_EXEC']
-
-    def maps(self):
-        for elf_segment in self.elf.iter_segments():
-            if elf_segment.header.p_type != 'PT_LOAD' or elf_segment.header.p_memsz == 0:
-                continue
-
-            flags = elf_segment.header.p_flags
-            # PF_X 0x1 Execute - PF_W 0x2 Write - PF_R 0x4 Read
-            perms = ['   ', '  x', ' w ', ' wx', 'r  ', 'r x', 'rw ', 'rwx'][flags & 7]
-            if 'r' not in perms:
-                raise Exception("Not readable map from cgc elf not supported")
-
-            # CGCMAP--
-            assert elf_segment.header.p_filesz != 0 or elf_segment.header.p_memsz != 0
-            yield((elf_segment.header.p_vaddr,
-                   elf_segment.header.p_memsz,
-                   perms,
-                   elf_segment.stream.name, elf_segment.header.p_offset, elf_segment.header.p_filesz))
-
-    def getInterpreter(self):
-        return self.interpreter
-
-    def threads(self):
-        yield(('Running', {'EIP': self.elf.header.e_entry}))
-
-
-Binary.magics = {b'\x7fCGC': CGCElf,
-                 b'\x7fELF': Elf}
+from .binary import Binary, CGCElf, Elf  # noqa
 
 
 if __name__ == '__main__':

manticore/binary/binary.py

@@ -0,0 +1,119 @@
+import io
+
+from elftools.elf.elffile import ELFFile
+
+
+class Binary:
+    magics = {}
+
+    def __new__(cls, path):
+        if cls is Binary:
+            cl = cls.magics[open(path, 'rb').read(4)]
+            return cl(path)
+        else:
+            return super(Binary, cls).__new__(cls)
+
+    def __init__(self, path):
+        self.path = path
+        self.magic = Binary.magics[open(path, 'rb').read(4)]
+
+    def arch(self):
+        pass
+
+    def maps(self):
+        pass
+
+    def threads(self):
+        pass
+
+
+class CGCElf(Binary):
+
+    @staticmethod
+    def _cgc2elf(filename):
+        # hack begin so we can use upstream Elftool
+        with open(filename, 'rb') as fd:
+            stream = io.BytesIO(fd.read())
+            stream.write(b'\x7fELF')
+            stream.name = fd.name
+            return stream
+
+    def __init__(self, filename):
+        super().__init__(filename)
+        stream = self._cgc2elf(filename)
+        self.elf = ELFFile(stream)
+        self.arch = {'x86': 'i386', 'x64': 'amd64'}[self.elf.get_machine_arch()]
+
+        assert 'i386' == self.arch
+        assert self.elf.header.e_type in ['ET_EXEC']
+
+    def maps(self):
+        for elf_segment in self.elf.iter_segments():
+            if elf_segment.header.p_type not in ['PT_LOAD', 'PT_NULL', 'PT_PHDR', 'PT_CGCPOV2']:
+                raise Exception("Not Supported Section")
+
+            if elf_segment.header.p_type != 'PT_LOAD' or elf_segment.header.p_memsz == 0:
+                continue
+
+            flags = elf_segment.header.p_flags
+            # PF_X 0x1 Execute - PF_W 0x2 Write - PF_R 0x4 Read
+            perms = ['   ', '  x', ' w ', ' wx', 'r  ', 'r x', 'rw ', 'rwx'][flags & 7]
+            if 'r' not in perms:
+                raise Exception("Not readable map from cgc elf not supported")
+
+            # CGCMAP--
+            assert elf_segment.header.p_filesz != 0 or elf_segment.header.p_memsz != 0
+            yield ((elf_segment.header.p_vaddr,
+                    elf_segment.header.p_memsz,
+                    perms,
+                    elf_segment.stream.name, elf_segment.header.p_offset, elf_segment.header.p_filesz))
+
+    def threads(self):
+        yield (('Running', {'EIP': self.elf.header.e_entry}))
+
+
+class Elf(Binary):
+    def __init__(self, filename):
+        super().__init__(filename)
+        self.elf = ELFFile(open(filename, 'rb'))
+        self.arch = {'x86': 'i386', 'x64': 'amd64'}[self.elf.get_machine_arch()]
+        assert self.elf.header.e_type in ['ET_DYN', 'ET_EXEC', 'ET_CORE']
+
+        # Get interpreter elf
+        self.interpreter = None
+        for elf_segment in self.elf.iter_segments():
+            if elf_segment.header.p_type != 'PT_INTERP':
+                continue
+            self.interpreter = Elf(elf_segment.data()[:-1])
+            break
+        if self.interpreter is not None:
+            assert self.interpreter.arch == self.arch
+            assert self.interpreter.elf.header.e_type in ['ET_DYN', 'ET_EXEC']
+
+    def maps(self):
+        for elf_segment in self.elf.iter_segments():
+            if elf_segment.header.p_type != 'PT_LOAD' or elf_segment.header.p_memsz == 0:
+                continue
+
+            flags = elf_segment.header.p_flags
+            # PF_X 0x1 Execute - PF_W 0x2 Write - PF_R 0x4 Read
+            perms = ['   ', '  x', ' w ', ' wx', 'r  ', 'r x', 'rw ', 'rwx'][flags & 7]
+            if 'r' not in perms:
+                raise Exception("Not readable map from cgc elf not supported")
+
+            # CGCMAP--
+            assert elf_segment.header.p_filesz != 0 or elf_segment.header.p_memsz != 0
+            yield ((elf_segment.header.p_vaddr,
+                    elf_segment.header.p_memsz,
+                    perms,
+                    elf_segment.stream.name, elf_segment.header.p_offset, elf_segment.header.p_filesz))
+
+    def getInterpreter(self):
+        return self.interpreter
+
+    def threads(self):
+        yield (('Running', {'EIP': self.elf.header.e_entry}))
+
+
+Binary.magics = {b'\x7fCGC': CGCElf,
+                 b'\x7fELF': Elf}

manticore/core/smtlib/__init__.py

@@ -6,18 +6,3 @@
 
 import logging
 logger = logging.getLogger(__name__)
-
-'''
-class OperationNotPermited(SolverException):
-    def __init__(self):
-        super().__init__("You cant build this expression")     #no childrens
-
-class ConcretizeException(SolverException):
-    def __init__(self, expression):
-        super().__init__("Need to concretize the following and retry\n"+str(expression))     #no childrens
-        self.expression = expression
-'''
-
-
-class VisitorException(Exception):
-    pass

manticore/core/smtlib/solver.py

@@ -23,7 +23,7 @@
 import shlex
 import time
 from .visitors import *
-from ...utils.helpers import issymbolic, istainted, taint_with, get_taints, memoized
+from ...utils.helpers import issymbolic, istainted, taint_with, get_taints
 import io
 import collections
 

manticore/platforms/decree.py

@@ -7,7 +7,7 @@
 from ..core.smtlib import *
 from ..core.executor import TerminateState
 from ..utils.helpers import issymbolic
-from ..binary import CGCElf
+from manticore.binary import CGCElf
 from ..platforms.platform import Platform
 import logging
 import random

manticore/platforms/evm.py

@@ -3,7 +3,7 @@
 import copy
 import inspect
 from functools import wraps
-from ..utils.helpers import issymbolic, memoized, get_taints, taint_with, istainted
+from ..utils.helpers import issymbolic, get_taints, taint_with, istainted
 from ..platforms.platform import *
 from ..core.smtlib import solver, BitVec, Array, Operators, Constant, ArrayVariable, BitVecConstant, translate_to_smtlib
 from ..core.state import Concretize, TerminateState

manticore/utils/helpers.py

@@ -72,38 +72,6 @@ def taint_with(arg, taint, value_bits=256, index_bits=256):
     return arg
 
 
-class memoized(object):
-    '''Decorator. Caches a function's return value each time it is called.
-    If called later with the same arguments, the cached value is returned
-    (not reevaluated).
-    '''
-
-    def __init__(self, func):
-        self.func = func
-        self.cache = {}
-
-    def __call__(self, *args, **kwargs):
-        key = args + tuple(sorted(kwargs.items()))
-        if not isinstance(key, collections.Hashable):
-            # uncacheable. a list, for instance.
-            # better to not cache than blow up.
-            return self.func(*args, **kwargs)
-        if key in self.cache:
-            return self.cache[key]
-        else:
-            value = self.func(*args, **kwargs)
-            self.cache[key] = value
-            return value
-
-    def __repr__(self):
-        '''Return the function's docstring.'''
-        return self.func.__doc__
-
-    def __get__(self, obj, objtype):
-        '''Support instance methods.'''
-        return functools.partial(self.__call__, obj)
-
-
 class CacheDict(OrderedDict):
     def __init__(self, *args, max_size=30000, flush_perc=30, **kwargs):
         self._max_size = max_size

tests/test_binaries.py

@@ -7,11 +7,37 @@
 import hashlib
 import subprocess
 import time
+from manticore.binary import Elf, CGCElf
 
 #logging.basicConfig(filename = "test.log",
 #                format = "%(asctime)s: %(name)s:%(levelname)s: %(message)s",
 #                level = logging.DEBUG)
 
+
+class TestBinaryPackage(unittest.TestCase):
+    _multiprocess_can_split_ = True
+
+    def test_elf(self):
+        filename = os.path.join(os.path.dirname(__file__), 'binaries', 'basic_linux_amd64')
+        f = Elf(filename)
+        self.assertTrue(
+            [(4194304, 823262, 'r x', 'tests/binaries/basic_linux_amd64', 0, 823262),
+             (7118520, 16112, 'rw ', 'tests/binaries/basic_linux_amd64', 827064, 7320)],
+            list(f.maps())
+        )
+        self.assertTrue([('Running', {'EIP': 4196624})], list(f.threads()))
+        self.assertIsNone(f.getInterpreter())
+
+    def test_decree(self):
+        filename = os.path.join(os.path.dirname(__file__), 'binaries', 'cadet_decree_x86')
+        f = CGCElf(filename)
+        self.assertTrue(
+            [(134512640, 1478, 'r x', 'tests/binaries/cadet_decree_x86', 0, 1478)],
+            list(f.maps())
+        )
+        self.assertTrue([('Running', {'EIP': 134513708})], list(f.threads()))
+
+
 class IntegrationTest(unittest.TestCase):
     _multiprocess_can_split_ = True
     def setUp(self):

tox.ini

@@ -4,7 +4,7 @@ envlist = py3{6,7}
 [testenv]
 deps = 
 	.[dev]
-commands = nosetests
+commands = nosetests tests
 install_command = pip install --no-binary keystone-engine {opts} {packages}
 
 [testenv:pep8]
@@ -20,4 +20,4 @@ exclude = docs/,examples/,scripts/,tests/
 [flake8]
 ignore = E265,E501,F403,F405,E266,E712,F841,E741,E722,E731
 max-line-length = 160
-exclude = .tox,.*.egg,.git,docs/,examples/,scripts/,tests/
+exclude = .tox,.*.egg,.git,docs/,examples/,scripts/,tests/,parsetab.py,iterpickle.py