Merge branch 'vispipeline'

angrutils/exploration.py

@@ -0,0 +1,26 @@
+import angr
+
+class NormalizedSteps(angr.exploration_techniques.ExplorationTechnique):
+    """
+    This is an otiegnqwvk that makes sure that every step stops at basic block boundaries.
+
+    Construct it with a normalized CFG.
+    """
+    def __init__(self, cfg):
+        super(NormalizedSteps, self).__init__()
+        self.cfg = cfg
+
+    def step(self, pg, stash, **kwargs):
+        kwargs['successor_func'] = self.normalized_step
+        return pg.step(stash=stash, **kwargs)
+
+    def normalized_step(self, path):
+        # cfg-acc doesn't normalize this, so we use the graph for now
+        # node = self.cfg.get_any_node(path.addr)
+        node = None
+        for n in self.cfg.nodes():
+            if n.addr == path.addr:
+                node = n
+                break
+        return path.step(num_inst=len(node.instruction_addrs) if node is not None else None)
+

angrutils/vis/__init__.py

@@ -0,0 +1,4 @@
+
+from .output import *
+from .base import *
+

angrutils/vis/angr/__init__.py

@@ -0,0 +1,7 @@
+
+
+from .source import *
+from .annotator import *
+from .content import *
+from .transform import *
+

angrutils/vis/angr/annotator.py

@@ -0,0 +1,137 @@
+
+from ..base import *
+
+import capstone
+
+class AngrColorSimprocedures(NodeAnnotator):
+    def __init__(self):
+        super(AngrColorSimprocedures, self).__init__()
+
+    def annotate_node(self, node):
+        if node.obj.is_simprocedure:
+            if node.obj.simprocedure_name in ['PathTerminator','ReturnUnconstrained','UnresolvableTarget']:
+                node.style = 'filled'
+                node.fillcolor = '#ffcccc'
+            else:
+                node.style = 'filled'
+                node.fillcolor = '#dddddd'
+
+class AngrColorExit(NodeAnnotator):
+    def __init__(self):
+        super(AngrColorExit, self).__init__()
+
+    def annotate_node(self, node):
+        if not node.obj.is_simprocedure:
+            found = False
+            for e in self.graph.edges:
+                if e.src == node:                
+                    found = True
+                    if 'jumpkind' in e.meta and e.meta['jumpkind'] == 'Ijk_Ret':
+                        node.style = 'filled'
+                        node.fillcolor = '#ddffdd'
+            if not found:
+                node.style = 'filled'
+                node.fillcolor = '#ddffdd'
+
+class AngrColorEntry(NodeAnnotator):
+    def __init__(self):
+        super(AngrColorEntry, self).__init__()
+
+    def annotate_node(self, node):
+        if not node.obj.is_simprocedure:
+            if hasattr(node.obj, 'function_address') and node.obj.addr == node.obj.function_address:
+                node.style = 'filled'
+                node.fillcolor = '#ffffcc'
+
+class AngrColorEdgesVex(EdgeAnnotator):
+    EDGECOLOR_CONDITIONAL_TRUE  = 'green'
+    EDGECOLOR_CONDITIONAL_FALSE = 'red'
+    EDGECOLOR_UNCONDITIONAL     = 'blue'
+    EDGECOLOR_CALL              = 'black'
+    EDGECOLOR_RET               = 'grey'
+    EDGECOLOR_UNKNOWN           = 'yellow'
+
+    def __init__(self):
+        super(AngrColorEdgesVex, self).__init__()
+
+
+    def annotate_edge(self, edge):
+        vex = None
+        if 'vex' in edge.src.content:
+            vex = edge.src.content['vex']['vex']
+
+        if 'jumpkind' in edge.meta:
+            jk = edge.meta['jumpkind']
+            if jk == 'Ijk_Ret':
+                edge.color = self.EDGECOLOR_RET
+            elif jk == 'Ijk_FakeRet':
+                edge.color = self.EDGECOLOR_RET
+                edge.style = 'dashed'
+            elif jk == 'Ijk_Call':
+                edge.color = self.EDGECOLOR_CALL
+                if len (vex.next.constants) == 1 and vex.next.constants[0].value != edge.dst.obj.addr:
+                    edge.style='dotted'
+            elif jk == 'Ijk_Boring':
+                if len(vex.constant_jump_targets) > 1:
+                    if len (vex.next.constants) == 1:
+                        if edge.dst.obj.addr == vex.next.constants[0].value:
+                            edge.color=self.EDGECOLOR_CONDITIONAL_FALSE
+                        else:
+                            edge.color=self.EDGECOLOR_CONDITIONAL_TRUE
+                    else:
+                        edge.color=self.EDGECOLOR_UNKNOWN
+                else:
+                    edge.color=self.EDGECOLOR_UNCONDITIONAL
+            else:
+                #TODO warning
+                edge.color = self.EDGECOLOR_UNKNOWN
+
+
+class AngrPathAnnotator(EdgeAnnotator, NodeAnnotator):
+
+    def __init__(self, path):
+        super(AngrPathAnnotator, self).__init__()
+        self.path = path
+        self.trace = list(path.addr_trace)
+
+    def set_graph(self, graph):
+        super(AngrPathAnnotator, self).set_graph(graph)
+        self.vaddr = self.valid_addrs()        
+        ftrace = filter(lambda _: _ in self.vaddr, self.trace)
+        self.edges_hit = set(zip(ftrace[:-1], ftrace[1:]))
+
+
+    def valid_addrs(self):
+        vaddr = set()
+        for n in self.graph.nodes:
+            vaddr.add(n.obj.addr)
+        return vaddr
+
+    #TODO add caching
+    #TODO not sure if this is valid
+    def node_hit(self, node):
+        ck = list(node.callstack_key)
+        ck.append(node.addr)
+        rtrace = list(reversed(self.trace))
+
+        found = True
+        si = 0
+        for c in reversed(ck):
+            if c == None:
+                break
+            try: 
+                si = rtrace[si:].index(c)
+            except:
+                found = False
+                break
+        return found
+
+    def annotate_edge(self, edge):
+        key = (edge.src.obj.addr, edge.dst.obj.addr)
+        if key in self.edges_hit:
+            edge.width = 3
+
+    def annotate_node(self, node):
+        if self.node_hit(node.obj):
+            node.width = 3
+

angrutils/vis/angr/content.py

@@ -0,0 +1,155 @@
+
+from ..base import *
+
+
+def safehex(val):
+    return str(hex(val) if val != None else None)
+
+class AngrCFGHead(Content):
+    def __init__(self):
+        super(AngrCFGHead, self).__init__('head', ['addr', 'func_addr', 'name', 'attributes'])
+
+    def gen_render(self, n):
+        node = n.obj
+        attributes=[]
+        if node.is_simprocedure:
+            attributes.append("SIMP")
+        if node.is_syscall:
+            attributes.append("SYSC")
+        if node.no_ret:
+            attributes.append("NORET")
+
+        label = "{:#08x} ({:#08x}) {} {}".format(node.addr, node.function_address, node.name, ' '.join(attributes))
+
+        n.content[self.name] = {
+            'data': [{
+                'addr': {
+                    'content': "{:#08x}".format(node.addr),
+                },
+                'func_addr' : {
+                    'content': "({:#08x})".format(node.function_address),
+                },
+                'name': {
+                    'content': node.name, 
+                    'style':'B'
+                },
+                'attributes': {
+                    'content': ' '.join(attributes)
+                }
+            }], 
+            'columns': self.get_columns()
+        }
+
+class AngrAsm(Content):
+    def __init__(self, project):
+        super(AngrAsm, self).__init__('asm', ['addr', 'mnemonic', 'operands'])
+        self.project = project        
+
+    def gen_render(self, n):
+        node = n.obj
+        if node.is_simprocedure or node.is_syscall:
+            return None
+
+        insns = self.project.factory.block(addr=node.addr, max_size=node.size).capstone.insns
+
+        data = []
+        for ins in insns:
+            data.append({
+                'addr': {
+                    'content': "0x%08x:\t" % ins.address,
+                    'align': 'LEFT'
+                },
+                'mnemonic': {
+                    'content': ins.mnemonic,
+                    'align': 'LEFT'
+                },
+                'operands': {
+                    'content': ins.op_str,
+                    'align': 'LEFT'
+                },
+                '_ins': ins
+            })
+
+        n.content[self.name] = {
+            'data': data,
+            'columns': self.get_columns(),
+        }
+
+
+class AngrVex(Content):
+    def __init__(self, project):
+        super(AngrVex, self).__init__('vex', ['addr', 'statement'])
+        self.project = project        
+
+    def gen_render(self, n):
+        node = n.obj
+        if node.is_simprocedure or node.is_syscall:
+            return None
+
+        vex = self.project.factory.block(addr=node.addr, max_size=node.size).vex
+
+        data = []
+        for j, s in enumerate(vex.statements):
+            data.append({
+                'addr': {
+                    'content': "0x%08x:" % j,
+                    'align': 'LEFT'
+                },
+                'statement': {
+                    'content': str(s),
+                    'align': 'LEFT'
+                },
+                '_stmt': s
+            })
+        data.append({
+            'addr': {
+                'content': "NEXT: ",
+                'align': 'LEFT'
+            },
+            'statement': {
+                'content': 'PUT(%s) = %s; %s' % (vex.arch.translate_register_name(vex.offsIP), vex.next, vex.jumpkind),
+                'align': 'LEFT'
+            }
+        })
+
+        n.content[self.name] = {
+            'data': data,
+            'columns': self.get_columns(),
+            'vex': vex
+        }
+
+class AngrCFGDebugInfo(Content):
+
+    def __init__(self):
+        super(AngrCFGDebugInfo, self).__init__('debug_info', ['text'])
+
+    def add_line(self, data, text):
+        data.append({
+            'text' : {
+                'align': 'LEFT',
+                'content' : text
+            }
+        })
+
+    def gen_render(self, n):
+        node = n.obj
+
+        data = []
+
+        self.add_line(data, "callstack_key: " + str([safehex(k) for k in node.callstack_key]))
+        self.add_line(data, "predecessors:")
+        for k in node.predecessors:
+            self.add_line(data, " - " + str(k))
+        self.add_line(data, "successors:")
+        for k in node.successors:
+            self.add_line(data, " - " + str(k))
+        self.add_line(data, "return_target: " + safehex(node.return_target))
+        self.add_line(data, "looping_times: " + str(node.looping_times))
+        self.add_line(data, "size: " + str(node.size))
+
+        n.content[self.name] = {
+            'data': data,
+            'columns': self.get_columns(),
+        }
+
+

angrutils/vis/angr/source.py

@@ -0,0 +1,31 @@
+
+from ..base import *
+
+import networkx
+
+class AngrCFGSource(Source):
+    def __init__(self):
+        super(AngrCFGSource, self).__init__()
+        self.lookup = {}
+        self.seq = 0
+
+    def parse(self, obj, graph):
+        if not isinstance(obj, networkx.classes.digraph.DiGraph):
+            raise VisError("Unknown input type '%s'" % type(obj))
+
+        for n in obj.nodes():
+            if n not in self.lookup:
+                wn = Node(self.seq, n)
+                self.seq += 1
+                self.lookup[n] = wn
+                graph.add_node(wn)
+            else:
+                raise VisError("Duplicate node %s" % str(n))
+
+        for src, dst, data in obj.edges(data=True):
+            if not src in self.lookup or not dst in self.lookup:
+                raise VisError("Missing nodes %s %s" % str(src), str(dst))
+            wsrc = self.lookup[src]
+            wdst = self.lookup[dst]
+            graph.add_edge(Edge(wsrc, wdst, data))
+