"lsp_processq_complete",
],
# zebra - main WQ
- ("mq_add_handler", "work_queue_add"): ["meta_queue_process",],
- ("meta_queue_process", "work_queue_add"): ["meta_queue_process",],
+ ("mq_add_handler", "work_queue_add"): [
+ "meta_queue_process",
+ ],
+ ("meta_queue_process", "work_queue_add"): [
+ "meta_queue_process",
+ ],
# bgpd - label pool WQ
- ("bgp_lp_get", "work_queue_add"): ["lp_cbq_docallback",],
- ("bgp_lp_event_chunk", "work_queue_add"): ["lp_cbq_docallback",],
- ("bgp_lp_event_zebra_up", "work_queue_add"): ["lp_cbq_docallback",],
+ ("bgp_lp_get", "work_queue_add"): [
+ "lp_cbq_docallback",
+ ],
+ ("bgp_lp_event_chunk", "work_queue_add"): [
+ "lp_cbq_docallback",
+ ],
+ ("bgp_lp_event_zebra_up", "work_queue_add"): [
+ "lp_cbq_docallback",
+ ],
# bgpd - main WQ
- ("bgp_process", "work_queue_add"): ["bgp_process_wq", "bgp_processq_del",],
- ("bgp_add_eoiu_mark", "work_queue_add"): ["bgp_process_wq", "bgp_processq_del",],
+ ("bgp_process", "work_queue_add"): [
+ "bgp_process_wq",
+ "bgp_processq_del",
+ ],
+ ("bgp_add_eoiu_mark", "work_queue_add"): [
+ "bgp_process_wq",
+ "bgp_processq_del",
+ ],
# clear node WQ
("bgp_clear_route_table", "work_queue_add"): [
"bgp_clear_route_node",
"bgp_clear_node_complete",
],
# rfapi WQs
- ("rfapi_close", "work_queue_add"): ["rfapi_deferred_close_workfunc",],
+ ("rfapi_close", "work_queue_add"): [
+ "rfapi_deferred_close_workfunc",
+ ],
("rfapiRibUpdatePendingNode", "work_queue_add"): [
"rfapiRibDoQueuedCallback",
"rfapiRibQueueItemDelete",
)
-frr_top_src = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+frr_top_src = os.path.dirname(
+ os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+)
+
def graph_iterate(graph):
"""iterator yielding all nodes of a graph
# with this program; see the file COPYING; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
-'''
+"""
Wrapping layer and additional utility around _clippy.ELFFile.
Essentially, the C bits have the low-level ELF access bits that should be
module (through GElf_*) and this code (cf. struct.unpack format mangling
in ELFDissectStruct) will take appropriate measures to flip and resize
fields as needed.
-'''
+"""
import struct
from collections import OrderedDict
# data access
#
+
class ELFNull(object):
- '''
+ """
NULL pointer, returned instead of ELFData
- '''
+ """
+
def __init__(self):
self.symname = None
self._dstsect = None
def __repr__(self):
- return '<ptr: NULL>'
+ return "<ptr: NULL>"
def __hash__(self):
return hash(None)
def get_string(self):
return None
+
class ELFUnresolved(object):
- '''
+ """
Reference to an unresolved external symbol, returned instead of ELFData
:param symname: name of the referenced symbol
:param addend: offset added to the symbol, normally zero
- '''
+ """
+
def __init__(self, symname, addend):
self.addend = addend
self.symname = symname
self._dstsect = None
def __repr__(self):
- return '<unresolved: %s+%d>' % (self.symname, self.addend)
+ return "<unresolved: %s+%d>" % (self.symname, self.addend)
def __hash__(self):
return hash((self.symname, self.addend))
+
class ELFData(object):
- '''
+ """
Actual data somewhere in the ELF file.
:type dstsect: ELFSubset
:param dstsect: container data area (section or entire file)
:param dstoffs: byte offset into dstsect
:param dstlen: byte size of object, or None if unknown, open-ended or string
- '''
+ """
+
def __init__(self, dstsect, dstoffs, dstlen):
self._dstsect = dstsect
self._dstoffs = dstoffs
self.symname = None
def __repr__(self):
- return '<ptr: %s+0x%05x/%d>' % (self._dstsect.name, self._dstoffs, self._dstlen or -1)
+ return "<ptr: %s+0x%05x/%d>" % (
+ self._dstsect.name,
+ self._dstoffs,
+ self._dstlen or -1,
+ )
def __hash__(self):
return hash((self._dstsect, self._dstoffs))
def get_string(self):
- '''
+ """
Interpret as C string / null terminated UTF-8 and get the actual text.
- '''
+ """
try:
- return self._dstsect[self._dstoffs:str].decode('UTF-8')
+ return self._dstsect[self._dstoffs : str].decode("UTF-8")
except:
- import pdb; pdb.set_trace()
+ import pdb
+
+ pdb.set_trace()
def get_data(self, reflen):
- '''
+ """
Interpret as some structure (and check vs. expected length)
:param reflen: expected size of the object, compared against actual
size (which is only known in rare cases, mostly when directly
accessing a symbol since symbols have their destination object
size recorded)
- '''
+ """
if self._dstlen is not None and self._dstlen != reflen:
- raise ValueError('symbol size mismatch (got %d, expected %d)' % (self._dstlen, reflen))
- return self._dstsect[self._dstoffs:self._dstoffs+reflen]
+ raise ValueError(
+ "symbol size mismatch (got %d, expected %d)" % (self._dstlen, reflen)
+ )
+ return self._dstsect[self._dstoffs : self._dstoffs + reflen]
def offset(self, offs, within_symbol=False):
- '''
+ """
Get another ELFData at an offset
:param offs: byte offset, can be negative (e.g. in container_of)
:param within_symbol: retain length information
- '''
+ """
if self._dstlen is None or not within_symbol:
return ELFData(self._dstsect, self._dstoffs + offs, None)
else:
return ELFData(self._dstsect, self._dstoffs + offs, self._dstlen - offs)
+
#
# dissection data items
#
+
class ELFDissectData(object):
- '''
+ """
Common bits for ELFDissectStruct and ELFDissectUnion
- '''
+ """
def __len__(self):
- '''
+ """
Used for boolean evaluation, e.g. "if struct: ..."
- '''
- return not (isinstance(self._data, ELFNull) or isinstance(self._data, ELFUnresolved))
+ """
+ return not (
+ isinstance(self._data, ELFNull) or isinstance(self._data, ELFUnresolved)
+ )
def container_of(self, parent, fieldname):
- '''
+ """
Assume this struct is embedded in a larger struct and get at the larger
Python ``self.container_of(a, b)`` = C ``container_of(self, a, b)``
:param parent: class (not instance) of the larger struct
:param fieldname: fieldname that refers back to this
:returns: instance of parent, with fieldname set to this object
- '''
+ """
offset = 0
- if not hasattr(parent, '_efields'):
+ if not hasattr(parent, "_efields"):
parent._setup_efields()
for field in parent._efields[self.elfclass]:
if field[0] == fieldname:
break
spec = field[1]
- if spec == 'P':
- spec = 'I' if self.elfclass == 32 else 'Q'
+ if spec == "P":
+ spec = "I" if self.elfclass == 32 else "Q"
offset += struct.calcsize(spec)
else:
- raise AttributeError('%r not found in %r.fields' % (fieldname, parent))
+ raise AttributeError("%r not found in %r.fields" % (fieldname, parent))
+
+ return parent(self._data.offset(-offset), replace={fieldname: self})
- return parent(self._data.offset(-offset), replace = {fieldname: self})
class ELFDissectStruct(ELFDissectData):
- '''
+ """
Decode and provide access to a struct somewhere in the ELF file
Handles pointers and strings somewhat nicely. Create a subclass for each
.. attribute:: fieldrename
Dictionary to rename fields, useful if fields comes from tiabwarfo.py.
- '''
+ """
class Pointer(object):
- '''
+ """
Quick wrapper for pointers to further structs
This is just here to avoid going into infinite loops when loading
structs that have pointers to each other (e.g. struct xref <-->
struct xrefdata.) The pointer destination is only instantiated when
actually accessed.
- '''
+ """
+
def __init__(self, cls, ptr):
self.cls = cls
self.ptr = ptr
def __repr__(self):
- return '<Pointer:%s %r>' % (self.cls.__name__, self.ptr)
+ return "<Pointer:%s %r>" % (self.cls.__name__, self.ptr)
def __call__(self):
if isinstance(self.ptr, ELFNull):
return None
return self.cls(self.ptr)
- def __new__(cls, dataptr, parent = None, replace = None):
+ def __new__(cls, dataptr, parent=None, replace=None):
if dataptr._dstsect is None:
return super().__new__(cls)
dataptr._dstsect._pointers[(cls, dataptr)] = obj
return obj
- replacements = 'lLnN'
+ replacements = "lLnN"
@classmethod
def _preproc_structspec(cls, elfclass, spec):
elfbits = elfclass
- if hasattr(spec, 'calcsize'):
- spec = '%ds' % (spec.calcsize(elfclass),)
+ if hasattr(spec, "calcsize"):
+ spec = "%ds" % (spec.calcsize(elfclass),)
if elfbits == 32:
- repl = ['i', 'I']
+ repl = ["i", "I"]
else:
- repl = ['q', 'Q']
+ repl = ["q", "Q"]
for c in cls.replacements:
spec = spec.replace(c, repl[int(c.isupper())])
return spec
size += struct.calcsize(newf[1])
cls._esize[elfclass] = size
- def __init__(self, dataptr, parent = None, replace = None):
- if not hasattr(self.__class__, '_efields'):
+ def __init__(self, dataptr, parent=None, replace=None):
+ if not hasattr(self.__class__, "_efields"):
self._setup_efields()
self._fdata = None
# need to correlate output from struct.unpack with extra metadata
# about the particular fields, so note down byte offsets (in locs)
# and tuple indices of pointers (in ptrs)
- pspec = ''
+ pspec = ""
locs = {}
ptrs = set()
for idx, spec in enumerate(pspecl):
- if spec == 'P':
+ if spec == "P":
ptrs.add(idx)
spec = self._elfsect.ptrtype
self._fdata[name] = replace[name]
continue
- if isinstance(self.fields[i][1], type) and issubclass(self.fields[i][1], ELFDissectData):
+ if isinstance(self.fields[i][1], type) and issubclass(
+ self.fields[i][1], ELFDissectData
+ ):
dataobj = self.fields[i][1](dataptr.offset(locs[i]), self)
self._fdata[name] = dataobj
continue
def __repr__(self):
if not isinstance(self._data, ELFData):
- return '<%s: %r>' % (self.__class__.__name__, self._data)
- return '<%s: %s>' % (self.__class__.__name__,
- ', '.join(['%s=%r' % t for t in self._fdata.items()]))
+ return "<%s: %r>" % (self.__class__.__name__, self._data)
+ return "<%s: %s>" % (
+ self.__class__.__name__,
+ ", ".join(["%s=%r" % t for t in self._fdata.items()]),
+ )
@classmethod
def calcsize(cls, elfclass):
- '''
+ """
Sum up byte size of this struct
Wraps struct.calcsize with some extra features.
- '''
- if not hasattr(cls, '_efields'):
+ """
+ if not hasattr(cls, "_efields"):
cls._setup_efields()
- pspec = ''.join([f[1] for f in cls._efields[elfclass]])
+ pspec = "".join([f[1] for f in cls._efields[elfclass]])
- ptrtype = 'I' if elfclass == 32 else 'Q'
- pspec = pspec.replace('P', ptrtype)
+ ptrtype = "I" if elfclass == 32 else "Q"
+ pspec = pspec.replace("P", ptrtype)
return struct.calcsize(pspec)
+
class ELFDissectUnion(ELFDissectData):
- '''
+ """
Decode multiple structs in the same place.
Not currently used (and hence not tested.) Worked at some point but not
needed anymore and may be borked now. Remove this comment when using.
- '''
- def __init__(self, dataptr, parent = None):
+ """
+
+ def __init__(self, dataptr, parent=None):
self._dataptr = dataptr
self._parent = parent
self.members = []
setattr(self, name, item)
def __repr__(self):
- return '<%s: %s>' % (self.__class__.__name__, ', '.join([repr(i) for i in self.members]))
+ return "<%s: %s>" % (
+ self.__class__.__name__,
+ ", ".join([repr(i) for i in self.members]),
+ )
@classmethod
def calcsize(cls, elfclass):
return max([member.calcsize(elfclass) for name, member in cls.members])
+
#
# wrappers for spans of ELF data
#
+
class ELFSubset(object):
- '''
+ """
Common abstract base for section-level and file-level access.
- '''
+ """
def __init__(self):
super().__init__()
return hash(self.name)
def __getitem__(self, k):
- '''
+ """
Read data from slice
Subscript **must** be a slice; a simple index will not return a byte
- `this[123:456]` - extract specific range
- `this[123:str]` - extract until null byte. The slice stop value is
the `str` type (or, technically, `unicode`.)
- '''
+ """
return self._obj[k]
def getreloc(self, offset):
- '''
+ """
Check for a relocation record at the specified offset.
- '''
+ """
return self._obj.getreloc(offset)
- def iter_data(self, scls, slice_ = slice(None)):
- '''
+ def iter_data(self, scls, slice_=slice(None)):
+ """
Assume an array of structs present at a particular slice and decode
:param scls: ELFDissectData subclass for the struct
:param slice_: optional range specification
- '''
+ """
size = scls.calcsize(self._elffile.elfclass)
offset = slice_.start or 0
offset += size
def pointer(self, offset):
- '''
+ """
Try to dereference a pointer value
This checks whether there's a relocation at the given offset and
:param offset: byte offset from beginning of section,
or virtual address in file
:returns: ELFData wrapping pointed-to object
- '''
+ """
ptrsize = struct.calcsize(self.ptrtype)
- data = struct.unpack(self.endian + self.ptrtype, self[offset:offset + ptrsize])[0]
+ data = struct.unpack(
+ self.endian + self.ptrtype, self[offset : offset + ptrsize]
+ )[0]
reloc = self.getreloc(offset)
dstsect = None
# wrap_data is different between file & section
return self._wrap_data(data, dstsect)
+
class ELFDissectSection(ELFSubset):
- '''
+ """
Access the contents of an ELF section like ``.text`` or ``.data``
:param elfwrap: ELFDissectFile wrapper for the file
:param idx: section index in section header table
:param section: section object from C module
- '''
+ """
def __init__(self, elfwrap, idx, section):
super().__init__()
dstsect = self._elfwrap.get_section(dstsect.idx)
return ELFData(dstsect, offs, None)
+
class ELFDissectFile(ELFSubset):
- '''
+ """
Access the contents of an ELF file.
Note that offsets for array subscript and relocation/pointer access are
address like 0x400000 on x86.
:param filename: ELF file to open
- '''
+ """
def __init__(self, filename):
super().__init__()
self._elffile = self._obj = ELFFile(filename)
self._sections = {}
- self.ptrtype = 'I' if self._elffile.elfclass == 32 else 'Q'
- self.endian = '>' if self._elffile.bigendian else '<'
+ self.ptrtype = "I" if self._elffile.elfclass == 32 else "Q"
+ self.endian = ">" if self._elffile.bigendian else "<"
@property
def _elfwrap(self):
return ELFData(self, data, None)
def get_section(self, secname):
- '''
+ """
Look up section by name or index
- '''
+ """
if isinstance(secname, int):
sh_idx = secname
section = self._elffile.get_section_idx(secname)
import struct
from hashlib import sha256
-def bititer(data, bits, startbit = True):
- '''
+
+def bititer(data, bits, startbit=True):
+ """
just iterate the individual bits out from a bytes object
if startbit is True, an '1' bit is inserted at the very beginning
goes <bits> at a time, starts at LSB.
- '''
+ """
bitavail, v = 0, 0
if startbit and len(data) > 0:
v = data.pop(0)
bitavail -= bits
v >>= bits
+
def base32c(data):
- '''
+ """
Crockford base32 with extra dashes
- '''
+ """
chs = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
- o = ''
+ o = ""
if type(data) == str:
data = [ord(v) for v in data]
else:
data = list(data)
for i, bits in enumerate(bititer(data, 5)):
if i == 5:
- o = o + '-'
+ o = o + "-"
elif i == 10:
break
o = o + chs[bits]
return o
+
def uidhash(filename, hashstr, hashu32a, hashu32b):
- '''
+ """
xref Unique ID hash used in FRRouting
- '''
- filename = '/'.join(filename.rsplit('/')[-2:])
+ """
+ filename = "/".join(filename.rsplit("/")[-2:])
- hdata = filename.encode('UTF-8') + hashstr.encode('UTF-8')
- hdata += struct.pack('>II', hashu32a, hashu32b)
+ hdata = filename.encode("UTF-8") + hashstr.encode("UTF-8")
+ hdata += struct.pack(">II", hashu32a, hashu32b)
i = sha256(hdata).digest()
return base32c(i)
# combine daemon .xref files into frr.xref
out_lines.append("")
xref_targets = [
- target for target in xref_targets if target not in [
+ target
+ for target in xref_targets
+ if target
+ not in [
"bgpd/rfp-example/rfptest/rfptest",
"pimd/mtracebis",
"tools/ssd",
try:
import _clippy
except ImportError:
- sys.stderr.write('''these tests need to be run with the _clippy C extension
+ sys.stderr.write(
+ """these tests need to be run with the _clippy C extension
module available. Try running "clippy runtests.py ...".
-''')
+"""
+ )
sys.exit(1)
os.chdir(os.path.dirname(os.path.abspath(__file__)))
from pprint import pprint
root = os.path.dirname(os.path.dirname(__file__))
-sys.path.append(os.path.join(root, 'python'))
+sys.path.append(os.path.join(root, "python"))
import xrelfo
from clippy import elf, uidhash
+
def test_uidhash():
- assert uidhash.uidhash("lib/test_xref.c", "logging call", 3, 0) \
- == 'H7KJB-67TBH'
+ assert uidhash.uidhash("lib/test_xref.c", "logging call", 3, 0) == "H7KJB-67TBH"
+
def test_xrelfo_other():
for data in [
- elf.ELFNull(),
- elf.ELFUnresolved('somesym', 0),
- ]:
+ elf.ELFNull(),
+ elf.ELFUnresolved("somesym", 0),
+ ]:
dissect = xrelfo.XrefPtr(data)
print(repr(dissect))
with pytest.raises(AttributeError):
dissect.xref
+
def test_xrelfo_obj():
xrelfo_ = xrelfo.Xrelfo()
- edf = xrelfo_.load_elf(os.path.join(root, 'lib/.libs/zclient.o'), 'zclient.lo')
+ edf = xrelfo_.load_elf(os.path.join(root, "lib/.libs/zclient.o"), "zclient.lo")
xrefs = xrelfo_._xrefs
with pytest.raises(elf.ELFAccessError):
pprint(xrefs[0])
pprint(xrefs[0]._data)
+
def test_xrelfo_bin():
xrelfo_ = xrelfo.Xrelfo()
- edf = xrelfo_.load_elf(os.path.join(root, 'lib/.libs/libfrr.so'), 'libfrr.la')
+ edf = xrelfo_.load_elf(os.path.join(root, "lib/.libs/libfrr.so"), "libfrr.la")
xrefs = xrelfo_._xrefs
- assert edf[0:4] == b'\x7fELF'
+ assert edf[0:4] == b"\x7fELF"
pprint(xrefs[0])
pprint(xrefs[0]._data)
import argparse
import json
-structs = ['xref', 'xref_logmsg', 'xref_threadsched', 'xref_install_element', 'xrefdata', 'xrefdata_logmsg', 'cmd_element']
-
-def extract(filename='lib/.libs/libfrr.so'):
- '''
+structs = [
+ "xref",
+ "xref_logmsg",
+ "xref_threadsched",
+ "xref_install_element",
+ "xrefdata",
+ "xrefdata_logmsg",
+ "cmd_element",
+]
+
+
+def extract(filename="lib/.libs/libfrr.so"):
+ """
Convert output from "pahole" to JSON.
Example pahole output:
/* size: 32, cachelines: 1, members: 5 */
/* last cacheline: 32 bytes */
};
- '''
- pahole = subprocess.check_output(['pahole', '-C', ','.join(structs), filename]).decode('UTF-8')
+ """
+ pahole = subprocess.check_output(
+ ["pahole", "-C", ",".join(structs), filename]
+ ).decode("UTF-8")
- struct_re = re.compile(r'^struct ([^ ]+) \{([^\}]+)};', flags=re.M | re.S)
- field_re = re.compile(r'^\s*(?P<type>[^;\(]+)\s+(?P<name>[^;\[\]]+)(?:\[(?P<array>\d+)\])?;\s*\/\*(?P<comment>.*)\*\/\s*$')
- comment_re = re.compile(r'^\s*\/\*.*\*\/\s*$')
+ struct_re = re.compile(r"^struct ([^ ]+) \{([^\}]+)};", flags=re.M | re.S)
+ field_re = re.compile(
+ r"^\s*(?P<type>[^;\(]+)\s+(?P<name>[^;\[\]]+)(?:\[(?P<array>\d+)\])?;\s*\/\*(?P<comment>.*)\*\/\s*$"
+ )
+ comment_re = re.compile(r"^\s*\/\*.*\*\/\s*$")
pastructs = struct_re.findall(pahole)
out = {}
for sname, data in pastructs:
this = out.setdefault(sname, {})
- fields = this.setdefault('fields', [])
+ fields = this.setdefault("fields", [])
lines = data.strip().splitlines()
next_offs = 0
for line in lines:
- if line.strip() == '':
+ if line.strip() == "":
continue
m = comment_re.match(line)
if m is not None:
m = field_re.match(line)
if m is not None:
- offs, size = m.group('comment').strip().split()
+ offs, size = m.group("comment").strip().split()
offs = int(offs)
size = int(size)
- typ_ = m.group('type').strip()
- name = m.group('name')
+ typ_ = m.group("type").strip()
+ name = m.group("name")
- if name.startswith('(*'):
+ if name.startswith("(*"):
# function pointer
- typ_ = typ_ + ' *'
- name = name[2:].split(')')[0]
+ typ_ = typ_ + " *"
+ name = name[2:].split(")")[0]
data = {
- 'name': name,
- 'type': typ_,
- # 'offset': offs,
- # 'size': size,
+ "name": name,
+ "type": typ_,
+ # 'offset': offs,
+ # 'size': size,
}
- if m.group('array'):
- data['array'] = int(m.group('array'))
+ if m.group("array"):
+ data["array"] = int(m.group("array"))
fields.append(data)
if offs != next_offs:
- raise ValueError('%d padding bytes before struct %s.%s' % (offs - next_offs, sname, name))
+ raise ValueError(
+ "%d padding bytes before struct %s.%s"
+ % (offs - next_offs, sname, name)
+ )
next_offs = offs + size
continue
- raise ValueError('cannot process line: %s' % line)
+ raise ValueError("cannot process line: %s" % line)
return out
+
class FieldApplicator(object):
- '''
+ """
Fill ELFDissectStruct fields list from pahole/JSON
Uses the JSON file created by the above code to fill in the struct fields
in subclasses of ELFDissectStruct.
- '''
+ """
# only what we really need. add more as needed.
packtypes = {
- 'int': 'i',
- 'uint8_t': 'B',
- 'uint16_t': 'H',
- 'uint32_t': 'I',
- 'char': 's',
+ "int": "i",
+ "uint8_t": "B",
+ "uint16_t": "H",
+ "uint32_t": "I",
+ "char": "s",
}
def __init__(self, data):
def resolve(self, cls):
out = []
- #offset = 0
+ # offset = 0
+
+ fieldrename = getattr(cls, "fieldrename", {})
- fieldrename = getattr(cls, 'fieldrename', {})
def mkname(n):
return (fieldrename.get(n, n),)
- for field in self.data[cls.struct]['fields']:
- typs = field['type'].split()
- typs = [i for i in typs if i not in ['const']]
+ for field in self.data[cls.struct]["fields"]:
+ typs = field["type"].split()
+ typs = [i for i in typs if i not in ["const"]]
# this will break reuse of xrefstructs.json across 32bit & 64bit
# platforms
- #if field['offset'] != offset:
+ # if field['offset'] != offset:
# assert offset < field['offset']
# out.append(('_pad', '%ds' % (field['offset'] - offset,)))
# pretty hacky C types handling, but covers what we need
ptrlevel = 0
- while typs[-1] == '*':
+ while typs[-1] == "*":
typs.pop(-1)
ptrlevel += 1
if ptrlevel > 0:
- packtype = ('P', None)
+ packtype = ("P", None)
if ptrlevel == 1:
- if typs[0] == 'char':
- packtype = ('P', str)
- elif typs[0] == 'struct' and typs[1] in self.clsmap:
- packtype = ('P', self.clsmap[typs[1]])
- elif typs[0] == 'enum':
- packtype = ('I',)
+ if typs[0] == "char":
+ packtype = ("P", str)
+ elif typs[0] == "struct" and typs[1] in self.clsmap:
+ packtype = ("P", self.clsmap[typs[1]])
+ elif typs[0] == "enum":
+ packtype = ("I",)
elif typs[0] in self.packtypes:
packtype = (self.packtypes[typs[0]],)
- elif typs[0] == 'struct':
+ elif typs[0] == "struct":
if typs[1] in self.clsmap:
packtype = (self.clsmap[typs[1]],)
else:
- raise ValueError('embedded struct %s not in extracted data' % (typs[1],))
+ raise ValueError(
+ "embedded struct %s not in extracted data" % (typs[1],)
+ )
else:
- raise ValueError('cannot decode field %s in struct %s (%s)' % (
- cls.struct, field['name'], field['type']))
-
- if 'array' in field and typs[0] == 'char':
- packtype = ('%ds' % field['array'],)
- out.append(mkname(field['name']) + packtype)
- elif 'array' in field:
- for i in range(0, field['array']):
- out.append(mkname('%s_%d' % (field['name'], i)) + packtype)
+ raise ValueError(
+ "cannot decode field %s in struct %s (%s)"
+ % (cls.struct, field["name"], field["type"])
+ )
+
+ if "array" in field and typs[0] == "char":
+ packtype = ("%ds" % field["array"],)
+ out.append(mkname(field["name"]) + packtype)
+ elif "array" in field:
+ for i in range(0, field["array"]):
+ out.append(mkname("%s_%d" % (field["name"], i)) + packtype)
else:
- out.append(mkname(field['name']) + packtype)
+ out.append(mkname(field["name"]) + packtype)
- #offset = field['offset'] + field['size']
+ # offset = field['offset'] + field['size']
cls.fields = out
for cls in self.classes:
self.resolve(cls)
+
def main():
- argp = argparse.ArgumentParser(description = 'FRR DWARF structure extractor')
- argp.add_argument('-o', dest='output', type=str, help='write JSON output', default='python/xrefstructs.json')
- argp.add_argument('-i', dest='input', type=str, help='ELF file to read', default='lib/.libs/libfrr.so')
+ argp = argparse.ArgumentParser(description="FRR DWARF structure extractor")
+ argp.add_argument(
+ "-o",
+ dest="output",
+ type=str,
+ help="write JSON output",
+ default="python/xrefstructs.json",
+ )
+ argp.add_argument(
+ "-i",
+ dest="input",
+ type=str,
+ help="ELF file to read",
+ default="lib/.libs/libfrr.so",
+ )
args = argp.parse_args()
out = extract(args.input)
- with open(args.output + '.tmp', 'w') as fd:
+ with open(args.output + ".tmp", "w") as fd:
json.dump(out, fd, indent=2, sort_keys=True)
- os.rename(args.output + '.tmp', args.output)
+ os.rename(args.output + ".tmp", args.output)
+
-if __name__ == '__main__':
+if __name__ == "__main__":
main()
from xref2vtysh import CommandEntry
try:
- with open(os.path.join(frr_top_src, 'python', 'xrefstructs.json'), 'r') as fd:
+ with open(os.path.join(frr_top_src, "python", "xrefstructs.json"), "r") as fd:
xrefstructs = json.load(fd)
except FileNotFoundError:
- sys.stderr.write('''
+ sys.stderr.write(
+ """
The "xrefstructs.json" file (created by running tiabwarfo.py with the pahole
tool available) could not be found. It should be included with the sources.
-''')
+"""
+ )
sys.exit(1)
# constants, need to be kept in sync manually...
# LOG_*
priovals = {}
-prios = ['0', '1', '2', 'E', 'W', 'N', 'I', 'D']
+prios = ["0", "1", "2", "E", "W", "N", "I", "D"]
class XrelfoJson(object):
def to_dict(self, refs):
pass
+
class Xref(ELFDissectStruct, XrelfoJson):
- struct = 'xref'
- fieldrename = {'type': 'typ'}
+ struct = "xref"
+ fieldrename = {"type": "typ"}
containers = {}
def __init__(self, *args, **kwargs):
def container(self):
if self._container is None:
if self.typ in self.containers:
- self._container = self.container_of(self.containers[self.typ], 'xref')
+ self._container = self.container_of(self.containers[self.typ], "xref")
return self._container
def check(self, *args, **kwargs):
class Xrefdata(ELFDissectStruct):
- struct = 'xrefdata'
+ struct = "xrefdata"
# uid is all zeroes in the data loaded from ELF
- fieldrename = {'uid': '_uid'}
+ fieldrename = {"uid": "_uid"}
def ref_from(self, xref, typ):
self.xref = xref
return None
return uidhash(self.xref.file, self.hashstr, self.hashu32_0, self.hashu32_1)
+
class XrefPtr(ELFDissectStruct):
fields = [
- ('xref', 'P', Xref),
+ ("xref", "P", Xref),
]
+
class XrefThreadSched(ELFDissectStruct, XrelfoJson):
- struct = 'xref_threadsched'
+ struct = "xref_threadsched"
+
+
Xref.containers[XREFT_THREADSCHED] = XrefThreadSched
+
class XrefLogmsg(ELFDissectStruct, XrelfoJson):
- struct = 'xref_logmsg'
+ struct = "xref_logmsg"
def _warn_fmt(self, text):
- lines = text.split('\n')
- yield ((self.xref.file, self.xref.line), '%s:%d: %s (in %s())%s\n' % (self.xref.file, self.xref.line, lines[0], self.xref.func, ''.join(['\n' + l for l in lines[1:]])))
+ lines = text.split("\n")
+ yield (
+ (self.xref.file, self.xref.line),
+ "%s:%d: %s (in %s())%s\n"
+ % (
+ self.xref.file,
+ self.xref.line,
+ lines[0],
+ self.xref.func,
+ "".join(["\n" + l for l in lines[1:]]),
+ ),
+ )
fmt_regexes = [
- (re.compile(r'([\n\t]+)'), 'error: log message contains tab or newline'),
- # (re.compile(r'^(\s+)'), 'warning: log message starts with whitespace'),
- (re.compile(r'^((?:warn(?:ing)?|error):\s*)', re.I), 'warning: log message starts with severity'),
+ (re.compile(r"([\n\t]+)"), "error: log message contains tab or newline"),
+ # (re.compile(r'^(\s+)'), 'warning: log message starts with whitespace'),
+ (
+ re.compile(r"^((?:warn(?:ing)?|error):\s*)", re.I),
+ "warning: log message starts with severity",
+ ),
]
arg_regexes = [
- # the (?<![\?:] ) avoids warning for x ? inet_ntop(...) : "(bla)"
- (re.compile(r'((?<![\?:] )inet_ntop\s*\(\s*(?:[AP]F_INET|2)\s*,)'), 'cleanup: replace inet_ntop(AF_INET, ...) with %pI4', lambda s: True),
- (re.compile(r'((?<![\?:] )inet_ntop\s*\(\s*(?:[AP]F_INET6|10)\s*,)'), 'cleanup: replace inet_ntop(AF_INET6, ...) with %pI6', lambda s: True),
- (re.compile(r'((?<![\?:] )inet_ntoa)'), 'cleanup: replace inet_ntoa(...) with %pI4', lambda s: True),
- (re.compile(r'((?<![\?:] )ipaddr2str)'), 'cleanup: replace ipaddr2str(...) with %pIA', lambda s: True),
- (re.compile(r'((?<![\?:] )prefix2str)'), 'cleanup: replace prefix2str(...) with %pFX', lambda s: True),
- (re.compile(r'((?<![\?:] )prefix_mac2str)'), 'cleanup: replace prefix_mac2str(...) with %pEA', lambda s: True),
- (re.compile(r'((?<![\?:] )sockunion2str)'), 'cleanup: replace sockunion2str(...) with %pSU', lambda s: True),
-
- # (re.compile(r'^(\s*__(?:func|FUNCTION|PRETTY_FUNCTION)__\s*)'), 'error: debug message starts with __func__', lambda s: (s.priority & 7 == 7) ),
+ # the (?<![\?:] ) avoids warning for x ? inet_ntop(...) : "(bla)"
+ (
+ re.compile(r"((?<![\?:] )inet_ntop\s*\(\s*(?:[AP]F_INET|2)\s*,)"),
+ "cleanup: replace inet_ntop(AF_INET, ...) with %pI4",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )inet_ntop\s*\(\s*(?:[AP]F_INET6|10)\s*,)"),
+ "cleanup: replace inet_ntop(AF_INET6, ...) with %pI6",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )inet_ntoa)"),
+ "cleanup: replace inet_ntoa(...) with %pI4",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )ipaddr2str)"),
+ "cleanup: replace ipaddr2str(...) with %pIA",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )prefix2str)"),
+ "cleanup: replace prefix2str(...) with %pFX",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )prefix_mac2str)"),
+ "cleanup: replace prefix_mac2str(...) with %pEA",
+ lambda s: True,
+ ),
+ (
+ re.compile(r"((?<![\?:] )sockunion2str)"),
+ "cleanup: replace sockunion2str(...) with %pSU",
+ lambda s: True,
+ ),
+ # (re.compile(r'^(\s*__(?:func|FUNCTION|PRETTY_FUNCTION)__\s*)'), 'error: debug message starts with __func__', lambda s: (s.priority & 7 == 7) ),
]
def check(self, wopt):
out = []
for i, text in enumerate(items):
if (i % 2) == 1:
- out.append('\033[41;37;1m%s\033[m' % repr(text)[1:-1])
+ out.append("\033[41;37;1m%s\033[m" % repr(text)[1:-1])
else:
out.append(repr(text)[1:-1])
- excerpt = ''.join(out)
+ excerpt = "".join(out)
else:
excerpt = repr(itext)[1:-1]
return excerpt
continue
excerpt = fmt_msg(rex, self.args)
- yield from self._warn_fmt('%s:\n\t"%s",\n\t%s' % (msg, repr(self.fmtstring)[1:-1], excerpt))
+ yield from self._warn_fmt(
+ '%s:\n\t"%s",\n\t%s' % (msg, repr(self.fmtstring)[1:-1], excerpt)
+ )
def dump(self):
- print('%-60s %s%s %-25s [EC %d] %s' % (
- '%s:%d %s()' % (self.xref.file, self.xref.line, self.xref.func),
- prios[self.priority & 7],
- priovals.get(self.priority & 0x30, ' '),
- self.xref.xrefdata.uid, self.ec, self.fmtstring))
+ print(
+ "%-60s %s%s %-25s [EC %d] %s"
+ % (
+ "%s:%d %s()" % (self.xref.file, self.xref.line, self.xref.func),
+ prios[self.priority & 7],
+ priovals.get(self.priority & 0x30, " "),
+ self.xref.xrefdata.uid,
+ self.ec,
+ self.fmtstring,
+ )
+ )
def to_dict(self, xrelfo):
- jsobj = dict([(i, getattr(self.xref, i)) for i in ['file', 'line', 'func']])
+ jsobj = dict([(i, getattr(self.xref, i)) for i in ["file", "line", "func"]])
if self.ec != 0:
- jsobj['ec'] = self.ec
- jsobj['fmtstring'] = self.fmtstring
- jsobj['args'] = self.args
- jsobj['priority'] = self.priority & 7
- jsobj['type'] = 'logmsg'
- jsobj['binary'] = self._elfsect._elfwrap.orig_filename
+ jsobj["ec"] = self.ec
+ jsobj["fmtstring"] = self.fmtstring
+ jsobj["args"] = self.args
+ jsobj["priority"] = self.priority & 7
+ jsobj["type"] = "logmsg"
+ jsobj["binary"] = self._elfsect._elfwrap.orig_filename
if self.priority & 0x10:
- jsobj.setdefault('flags', []).append('errno')
+ jsobj.setdefault("flags", []).append("errno")
if self.priority & 0x20:
- jsobj.setdefault('flags', []).append('getaddrinfo')
+ jsobj.setdefault("flags", []).append("getaddrinfo")
+
+ xrelfo["refs"].setdefault(self.xref.xrefdata.uid, []).append(jsobj)
- xrelfo['refs'].setdefault(self.xref.xrefdata.uid, []).append(jsobj)
Xref.containers[XREFT_LOGMSG] = XrefLogmsg
+
class CmdElement(ELFDissectStruct, XrelfoJson):
- struct = 'cmd_element'
+ struct = "cmd_element"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def to_dict(self, xrelfo):
- jsobj = xrelfo['cli'].setdefault(self.name, {}).setdefault(self._elfsect._elfwrap.orig_filename, {})
-
- jsobj.update({
- 'string': self.string,
- 'doc': self.doc,
- })
+ jsobj = (
+ xrelfo["cli"]
+ .setdefault(self.name, {})
+ .setdefault(self._elfsect._elfwrap.orig_filename, {})
+ )
+
+ jsobj.update(
+ {
+ "string": self.string,
+ "doc": self.doc,
+ }
+ )
if self.attr:
- jsobj['attr'] = attr = self.attr
+ jsobj["attr"] = attr = self.attr
for attrname in CmdAttr.__members__:
val = CmdAttr[attrname]
if attr & val:
- jsobj.setdefault('attrs', []).append(attrname.lower())
+ jsobj.setdefault("attrs", []).append(attrname.lower())
attr &= ~val
- jsobj['defun'] = dict([(i, getattr(self.xref, i)) for i in ['file', 'line', 'func']])
+ jsobj["defun"] = dict(
+ [(i, getattr(self.xref, i)) for i in ["file", "line", "func"]]
+ )
+
Xref.containers[XREFT_DEFUN] = CmdElement
+
class XrefInstallElement(ELFDissectStruct, XrelfoJson):
- struct = 'xref_install_element'
+ struct = "xref_install_element"
def to_dict(self, xrelfo):
- jsobj = xrelfo['cli'].setdefault(self.cmd_element.name, {}).setdefault(self._elfsect._elfwrap.orig_filename, {})
- nodes = jsobj.setdefault('nodes', [])
+ jsobj = (
+ xrelfo["cli"]
+ .setdefault(self.cmd_element.name, {})
+ .setdefault(self._elfsect._elfwrap.orig_filename, {})
+ )
+ nodes = jsobj.setdefault("nodes", [])
+
+ nodes.append(
+ {
+ "node": self.node_type,
+ "install": dict(
+ [(i, getattr(self.xref, i)) for i in ["file", "line", "func"]]
+ ),
+ }
+ )
- nodes.append({
- 'node': self.node_type,
- 'install': dict([(i, getattr(self.xref, i)) for i in ['file', 'line', 'func']]),
- })
Xref.containers[XREFT_INSTALL_ELEMENT] = XrefInstallElement
class Xrelfo(dict):
def __init__(self):
- super().__init__({
- 'refs': {},
- 'cli': {},
- })
+ super().__init__(
+ {
+ "refs": {},
+ "cli": {},
+ }
+ )
self._xrefs = []
def load_file(self, filename):
orig_filename = filename
- if filename.endswith('.la') or filename.endswith('.lo'):
- with open(filename, 'r') as fd:
+ if filename.endswith(".la") or filename.endswith(".lo"):
+ with open(filename, "r") as fd:
for line in fd:
line = line.strip()
- if line.startswith('#') or line == '' or '=' not in line:
+ if line.startswith("#") or line == "" or "=" not in line:
continue
- var, val = line.split('=', 1)
- if var not in ['library_names', 'pic_object']:
+ var, val = line.split("=", 1)
+ if var not in ["library_names", "pic_object"]:
continue
if val.startswith("'") or val.startswith('"'):
val = val[1:-1]
- if var == 'pic_object':
+ if var == "pic_object":
filename = os.path.join(os.path.dirname(filename), val)
break
val = val.strip().split()[0]
- filename = os.path.join(os.path.dirname(filename), '.libs', val)
+ filename = os.path.join(os.path.dirname(filename), ".libs", val)
break
else:
- raise ValueError('could not process libtool file "%s"' % orig_filename)
+ raise ValueError(
+ 'could not process libtool file "%s"' % orig_filename
+ )
while True:
- with open(filename, 'rb') as fd:
+ with open(filename, "rb") as fd:
hdr = fd.read(4)
- if hdr == b'\x7fELF':
+ if hdr == b"\x7fELF":
self.load_elf(filename, orig_filename)
return
- if hdr[:2] == b'#!':
+ if hdr[:2] == b"#!":
path, name = os.path.split(filename)
- filename = os.path.join(path, '.libs', name)
+ filename = os.path.join(path, ".libs", name)
continue
- if hdr[:1] == b'{':
- with open(filename, 'r') as fd:
+ if hdr[:1] == b"{":
+ with open(filename, "r") as fd:
self.load_json(fd)
return
- raise ValueError('cannot determine file type for %s' % (filename))
+ raise ValueError("cannot determine file type for %s" % (filename))
def load_elf(self, filename, orig_filename):
edf = ELFDissectFile(filename)
edf.orig_filename = orig_filename
- note = edf._elffile.find_note('FRRouting', 'XREF')
+ note = edf._elffile.find_note("FRRouting", "XREF")
if note is not None:
- endian = '>' if edf._elffile.bigendian else '<'
+ endian = ">" if edf._elffile.bigendian else "<"
mem = edf._elffile[note]
if edf._elffile.elfclass == 64:
- start, end = struct.unpack(endian + 'QQ', mem)
+ start, end = struct.unpack(endian + "QQ", mem)
start += note.start
end += note.start + 8
else:
- start, end = struct.unpack(endian + 'II', mem)
+ start, end = struct.unpack(endian + "II", mem)
start += note.start
end += note.start + 4
ptrs = edf.iter_data(XrefPtr, slice(start, end))
else:
- xrefarray = edf.get_section('xref_array')
+ xrefarray = edf.get_section("xref_array")
if xrefarray is None:
- raise ValueError('file has neither xref note nor xref_array section')
+ raise ValueError("file has neither xref note nor xref_array section")
ptrs = xrefarray.iter_data(XrefPtr)
for ptr in ptrs:
if ptr.xref is None:
- print('NULL xref')
+ print("NULL xref")
continue
self._xrefs.append(ptr.xref)
def load_json(self, fd):
data = json.load(fd)
- for uid, items in data['refs'].items():
- myitems = self['refs'].setdefault(uid, [])
+ for uid, items in data["refs"].items():
+ myitems = self["refs"].setdefault(uid, [])
for item in items:
if item in myitems:
continue
myitems.append(item)
- for cmd, items in data['cli'].items():
- self['cli'].setdefault(cmd, {}).update(items)
+ for cmd, items in data["cli"].items():
+ self["cli"].setdefault(cmd, {}).update(items)
return data
for xref in self._xrefs:
yield from xref.check(checks)
+
def main():
- argp = argparse.ArgumentParser(description = 'FRR xref ELF extractor')
- argp.add_argument('-o', dest='output', type=str, help='write JSON output')
- argp.add_argument('--out-by-file', type=str, help='write by-file JSON output')
- argp.add_argument('-c', dest='vtysh_cmds', type=str, help='write vtysh_cmd.c')
- argp.add_argument('-Wlog-format', action='store_const', const=True)
- argp.add_argument('-Wlog-args', action='store_const', const=True)
- argp.add_argument('-Werror', action='store_const', const=True)
- argp.add_argument('--profile', action='store_const', const=True)
- argp.add_argument('binaries', metavar='BINARY', nargs='+', type=str, help='files to read (ELF files or libtool objects)')
+ argp = argparse.ArgumentParser(description="FRR xref ELF extractor")
+ argp.add_argument("-o", dest="output", type=str, help="write JSON output")
+ argp.add_argument("--out-by-file", type=str, help="write by-file JSON output")
+ argp.add_argument("-c", dest="vtysh_cmds", type=str, help="write vtysh_cmd.c")
+ argp.add_argument("-Wlog-format", action="store_const", const=True)
+ argp.add_argument("-Wlog-args", action="store_const", const=True)
+ argp.add_argument("-Werror", action="store_const", const=True)
+ argp.add_argument("--profile", action="store_const", const=True)
+ argp.add_argument(
+ "binaries",
+ metavar="BINARY",
+ nargs="+",
+ type=str,
+ help="files to read (ELF files or libtool objects)",
+ )
args = argp.parse_args()
if args.profile:
import cProfile
- cProfile.runctx('_main(args)', globals(), {'args': args}, sort='cumtime')
+
+ cProfile.runctx("_main(args)", globals(), {"args": args}, sort="cumtime")
else:
_main(args)
+
def _main(args):
errors = 0
xrelfo = Xrelfo()
xrelfo.load_file(fn)
except:
errors += 1
- sys.stderr.write('while processing %s:\n' % (fn))
+ sys.stderr.write("while processing %s:\n" % (fn))
traceback.print_exc()
for option in dir(args):
- if option.startswith('W') and option != 'Werror':
+ if option.startswith("W") and option != "Werror":
checks = sorted(xrelfo.check(args))
- sys.stderr.write(''.join([c[-1] for c in checks]))
+ sys.stderr.write("".join([c[-1] for c in checks]))
if args.Werror and len(checks) > 0:
errors += 1
break
-
- refs = xrelfo['refs']
+ refs = xrelfo["refs"]
counts = {}
for k, v in refs.items():
- strs = set([i['fmtstring'] for i in v])
+ strs = set([i["fmtstring"] for i in v])
if len(strs) != 1:
- print('\033[31;1m%s\033[m' % k)
+ print("\033[31;1m%s\033[m" % k)
counts[k] = len(v)
out = xrelfo
outbyfile = {}
for uid, locs in refs.items():
for loc in locs:
- filearray = outbyfile.setdefault(loc['file'], [])
+ filearray = outbyfile.setdefault(loc["file"], [])
loc = dict(loc)
- del loc['file']
+ del loc["file"]
filearray.append(loc)
for k in outbyfile.keys():
- outbyfile[k] = sorted(outbyfile[k], key=lambda x: x['line'])
+ outbyfile[k] = sorted(outbyfile[k], key=lambda x: x["line"])
if errors:
sys.exit(1)
if args.output:
- with open(args.output + '.tmp', 'w') as fd:
+ with open(args.output + ".tmp", "w") as fd:
json.dump(out, fd, indent=2, sort_keys=True, **json_dump_args)
- os.rename(args.output + '.tmp', args.output)
+ os.rename(args.output + ".tmp", args.output)
if args.out_by_file:
- with open(args.out_by_file + '.tmp', 'w') as fd:
+ with open(args.out_by_file + ".tmp", "w") as fd:
json.dump(outbyfile, fd, indent=2, sort_keys=True, **json_dump_args)
- os.rename(args.out_by_file + '.tmp', args.out_by_file)
+ os.rename(args.out_by_file + ".tmp", args.out_by_file)
if args.vtysh_cmds:
- with open(args.vtysh_cmds + '.tmp', 'w') as fd:
+ with open(args.vtysh_cmds + ".tmp", "w") as fd:
CommandEntry.run(out, fd)
- os.rename(args.vtysh_cmds + '.tmp', args.vtysh_cmds)
+ os.rename(args.vtysh_cmds + ".tmp", args.vtysh_cmds)
if args.Werror and CommandEntry.warn_counter:
sys.exit(1)
-if __name__ == '__main__':
+if __name__ == "__main__":
main()
projects / mirror_frr.git / commitdiff