BinaryNinja plugin that import DWARF info using addr2line

ImportDwarf.py

import subprocess
import sys
import os
import re

from binaryninja import BackgroundTaskThread, PluginCommand
from binaryninja.interaction import get_directory_name_input


def print_err(*msg):
    sys.stderr.write(" ".join(map(str, msg)) + "\n")


class Addr2linePopulator(BackgroundTaskThread):
    def __init__(self, bv, code_path):
        BackgroundTaskThread.__init__(self, "Adding DWARF info with addr2line", False)
        self.bv        = bv
        self.code_path = code_path
        self.mapping_cache = dict()
        self.file_cache    = dict()

        # CGC fix
        f = open(self.bv.file.filename, "rb")
        if f.read(4) == b"\x7fCGC":
            data = f.read()
            data = b"\x7fELF" + data
            with open("/dev/shm/bin.elf", "wb") as fout:
                fout.write(data)
            self.bin = "/dev/shm/bin.elf"
        else:
            self.bin = self.bv.file.filename
        f.close()

    def update_progress(self, msg, curr, total):
        self.progress = f"Addr2linePopulator [{msg}] {curr} / {total}"

    def run_addr2line(self, addresses: list):
        proc = subprocess.Popen(
            ['addr2line', '-e', self.bin], stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = proc.communicate(input="\n".join(map(hex, addresses)).encode())
        output = stdout.decode("ASCII")

        prev_line = ""
        res       = dict()
        for addr, line in zip(addresses, output.split("\n")):
            if line == prev_line:
                continue
            prev_line = line

            path, linenum = line.split(":")
            if path == "??" or linenum == "?":
                continue

            res[addr] = (path, int(linenum))

        return res

    def get_local_path(self, path):
        if path in self.mapping_cache:
            return self.mapping_cache[path]

        path_splitted = path.split("/")
        i = len(path_splitted) - 1
        while i >= 0:
            subpath = "/".join(path_splitted[i:])
            guess_path = os.path.join(self.code_path, subpath)
            if os.path.isfile(guess_path):
                self.mapping_cache[path] = guess_path
                return guess_path
            i -= 1
        return None

    def get_row(self, path, i):
        i -= 1

        if path in self.file_cache:
            return self.file_cache[path][i]

        data = list()
        filename = os.path.basename(path)
        with open(path, "r") as fin:
            for linenum, line in enumerate(fin):
                line = line.strip()
                line = re.sub("^ +", "", line)
                data.append("%s:%d\t: %s" % (filename, linenum, line))

        self.file_cache[path] = data
        return data[i]

    def run_on_function(self, fun):
        addresses = list()
        for bb in fun.basic_blocks:
            for instr in bb.disassembly_text:
                addr = instr.address
                addresses.append(addr)

        mappings = self.run_addr2line(addresses)
        for addr in mappings:
            path, linenum = mappings[addr]
            local_path = self.get_local_path(path)
            if local_path is None:
                print_err("Unable to find", path)
                return False

            row = self.get_row(local_path, linenum)
            self.bv.set_comment_at(addr, row)
        return True

    def run(self):
        self.bv.update_analysis_and_wait()
        n_functions = len(self.bv.functions)

        for i, fun in enumerate(self.bv.functions):
            self.update_progress(fun.name, i+1, n_functions)
            if not self.run_on_function(fun):
                return


def addr2line_populator(bv):
    code_path = get_directory_name_input("Select source folder")
    if code_path is None:
        return

    task = Addr2linePopulator(bv, code_path)
    task.start()


PluginCommand.register(
    "Import DWARF info",
    "",
    addr2line_populator
)