# -*- coding: utf-8 -*- """ ------------------------------------------------- File Name： objelf Description : Author : liaozhaoyan date： 2022/11/5 ------------------------------------------------- Change Activity: 2022/11/5: ------------------------------------------------- """ __author__ = 'liaozhaoyan' import datetime import os import sys import shlex import time import json import gc import re from subprocess import PIPE, Popen from threading import Thread, Lock from atobj import CatObj from dict2db import Cobj2db class Cobj2json(object): reCellHead = re.compile(r"\<\d+\>\<[0-9a-f]+\>\:") reInParenthesis = re.compile(r"(?<=\()[^\(\)]+(?=\))") def __init__(self): super(Cobj2json, self).__init__() self._atObj = CatObj() self._getStr = None def _compileUnitHead(self, lines): while True: for i, line in enumerate(lines[:-1]): if line.startswith("Contents of the") and not line.startswith("Contents of the .debug_info section:"): return 0, None if Cobj2json.reCellHead.search(line): return 1, lines[i:] tail = lines[-1] s = tail + self._getStr() lines = s.split("\n") def _compileUnitEnd(self, lines): res = [] while True: for i, line in enumerate(lines[:-1]): line = line.lstrip() if len(line): if line[0] == "<": res.append(line) else: return res, lines[i:] tail = lines[-1] s = tail + self._getStr() lines = s.split("\n") def _findHead(self): tail = "" while True: s = tail + self._getStr() lines = s.split("\n") for i, line in enumerate(lines[:-1]): if line.startswith("Contents of the .debug_info section:"): return lines[i:] tail = lines[-1] def _parseLine(self, line): if Cobj2json.reCellHead.search(line): head, content = line.split(":", 1) levels, offsets = head.split("><") level = int(levels[1:], 10) offset = int(offsets[:-1], 16) tag = "tag_name" fit = Cobj2json.reInParenthesis.findall(content) if len(fit): value = fit[0] else: value = None else: level = -1 head, content = line.split(">", 1) offset = int(head[1:], 16) tags, values = content.split(":", 1) tag = tags.strip() if tag in self._atObj.atDict: values = values.lstrip() value = self._atObj.parse(tag, values) elif tag == "Unknown AT value": tag = None value = "" else: print("\t tag: %s not in at dict. value: %s", tag, values) value = values.strip() return level, offset, tag, value def _decode(self, lines): lastLevel = 0 dStack = [] dRet = {} for i, line in enumerate(lines): level, offset, tag, value = self._parseLine(line) if tag is None: # for Unknown AT value continue if level >= 0: if value is None: continue dRet = {tag: value, "offset": offset} if level == lastLevel: if level == 0: dStack.append(dRet) else: dParent = dStack[level - 1] dRet = {tag: value, "offset": offset} if "child" not in dParent: raise ValueError("should add child.") else: dParent['child'].append(dRet) dStack[level] = dRet elif level > lastLevel: # > need add child. if level > lastLevel + 1: print("pid %d" % os.getpid()) raise ValueError("level %d, lastLevel %d" % (level, lastLevel)) dParent = dStack[level - 1] if "child" in dParent: raise ValueError("should no child.") dParent['child'] = [dRet] dStack.append(dRet) lastLevel = level elif level < lastLevel: # back to last level dParent = dStack[level - 1] dParent['child'].append(dRet) dStack = dStack[:level+1] dStack[level] = dRet lastLevel = level else: # single line. dRet[tag] = value return dStack[0] def accept(self, getStr): self._getStr = getStr tail = self._findHead() flag, tail = self._compileUnitHead(tail) while flag: lines, tail = self._compileUnitEnd(tail) # with open("cell.txt", "w") as f: # f.write("\n".join(lines)) yield self._decode(lines) flag, tail = self._compileUnitHead(tail) class Ctest(object): def __init__(self, path="/root/1ext/code/surftrace/tools/dwarf/walks/samelf/dwarf.txt"): super(Ctest, self).__init__() self._handle = open(path, 'r') self._db = Cobj2db("elf", "info.db") def read(self): res = self._handle.read(1024) if res == "": raise OSError("end of file") return res def test(self): o = Cobj2json() ana = o.accept(self.read) for res in ana: self._db.walks(res) del res gc.collect() class CmonThread(Thread): def __init__(self, mon): super(CmonThread, self).__init__() self.daemon = True self._mon = mon self.start() def run(self): while True: time.sleep(0.1) if self._mon.poll() is not None: self._mon.closePipe() break class CobjElf(object): def __init__(self, elf): super(CobjElf, self).__init__() if not os.path.exists(elf) or not os.path.isfile(elf): raise ValueError("%s is not a file" % elf) self._l = Lock() self._r, self._w = -1, -1 self._elf = elf self._p = None def __del__(self): if hasattr(self, "_p"): self.closePipe() def _exec(self): cmd = "readelf -w %s" % self._elf with self._l: self._r, self._w = os.pipe() self._p = Popen(shlex.split(cmd), stdout=self._w) return self._p def poll(self): res = 0 with self._l: if hasattr(self, "_p"): res = self._p.poll() return res def closePipe(self): with self._l: if self._r >= 0: os.close(self._r) os.close(self._w) self._r = -1 self._w = -1 try: self._p.terminate() except OSError: pass del self._p def _read(self, size=16384): if sys.version_info.major == 2: res = os.read(self._r, size) else: res = os.read(self._r, size).decode() if res is None or res == "": raise OSError("end of file.") return res def out(self, toFile): self._exec() with open(toFile, 'w') as f: while True: try: s = self._read() except OSError: break f.write(s) def toDb(self, name, db): o = Cobj2json() self._exec() t = CmonThread(self) ana = o.accept(self._read) for res in ana: obj = Cobj2db(name, db) # with open("vm.json", "w") as f: # json.dump(res, f) obj.walks(res) self.closePipe() t.join(0.1) if __name__ == "__main__": e = CobjElf("/home/lbc/.output/lbc.bpf.o") # e.out("vm.db") t1 = datetime.datetime.now() e.toDb("bpf", "bpf.db") print("to Db.", datetime.datetime.now() - t1) # t = Ctest() # t = Ctest("./vm.txt") # t.test() pass