Something went wrong. Try again.
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254# SPDX-License-Identifier: GPL-2.0
import jsonimport subprocessimport tempfile
import gdb
from linux import constants, lists, radixtree, utils
if constants.LX_CONFIG_BPF and constants.LX_CONFIG_BPF_JIT: bpf_ksym_type = utils.CachedType("struct bpf_ksym")if constants.LX_CONFIG_BPF_SYSCALL: bpf_prog_type = utils.CachedType("struct bpf_prog")
def get_ksym_name(ksym): name = ksym["name"].bytes end = name.find(b"\x00") if end != -1: name = name[:end] return name.decode()
def list_ksyms(): if not (constants.LX_CONFIG_BPF and constants.LX_CONFIG_BPF_JIT): return [] bpf_kallsyms = gdb.parse_and_eval("&bpf_kallsyms") bpf_ksym_ptr_type = bpf_ksym_type.get_type().pointer() return list(lists.list_for_each_entry(bpf_kallsyms, bpf_ksym_ptr_type, "lnode"))
class KsymAddBreakpoint(gdb.Breakpoint): def __init__(self, monitor): super(KsymAddBreakpoint, self).__init__("bpf_ksym_add", internal=True) self.silent = True self.monitor = monitor
def stop(self): self.monitor.add(gdb.parse_and_eval("ksym")) return False
class KsymRemoveBreakpoint(gdb.Breakpoint): def __init__(self, monitor): super(KsymRemoveBreakpoint, self).__init__("bpf_ksym_del", internal=True) self.silent = True self.monitor = monitor
def stop(self): self.monitor.remove(gdb.parse_and_eval("ksym")) return False
class KsymMonitor: def __init__(self, add, remove): self.add = add self.remove = remove
self.add_bp = KsymAddBreakpoint(self) self.remove_bp = KsymRemoveBreakpoint(self)
self.notify_initial()
def notify_initial(self): for ksym in list_ksyms(): self.add(ksym)
def delete(self): self.add_bp.delete() self.remove_bp.delete()
def list_progs(): if not constants.LX_CONFIG_BPF_SYSCALL: return [] idr_rt = gdb.parse_and_eval("&prog_idr.idr_rt") bpf_prog_ptr_type = bpf_prog_type.get_type().pointer() progs = [] for _, slot in radixtree.for_each_slot(idr_rt): prog = slot.dereference().cast(bpf_prog_ptr_type) progs.append(prog) # Subprogs are not registered in prog_idr, fetch them manually. # func[0] is the current prog. aux = prog["aux"] func = aux["func"] real_func_cnt = int(aux["real_func_cnt"]) for i in range(1, real_func_cnt): progs.append(func[i]) return progs
class ProgAddBreakpoint(gdb.Breakpoint): def __init__(self, monitor): super(ProgAddBreakpoint, self).__init__("bpf_prog_kallsyms_add", internal=True) self.silent = True self.monitor = monitor
def stop(self): self.monitor.add(gdb.parse_and_eval("fp")) return False
class ProgRemoveBreakpoint(gdb.Breakpoint): def __init__(self, monitor): super(ProgRemoveBreakpoint, self).__init__("bpf_prog_free_id", internal=True) self.silent = True self.monitor = monitor
def stop(self): self.monitor.remove(gdb.parse_and_eval("prog")) return False
class ProgMonitor: def __init__(self, add, remove): self.add = add self.remove = remove
self.add_bp = ProgAddBreakpoint(self) self.remove_bp = ProgRemoveBreakpoint(self)
self.notify_initial()
def notify_initial(self): for prog in list_progs(): self.add(prog)
def delete(self): self.add_bp.delete() self.remove_bp.delete()
def btf_str_by_offset(btf, offset): while offset < btf["start_str_off"]: btf = btf["base_btf"]
offset -= btf["start_str_off"] if offset < btf["hdr"]["str_len"]: return (btf["strings"] + offset).string()
return None
def bpf_line_info_line_num(line_col): return line_col >> 10
def bpf_line_info_line_col(line_col): return line_col & 0x3ff
class LInfoIter: def __init__(self, prog): # See bpf_prog_get_file_line() for details. self.pos = 0 self.nr_linfo = 0
if prog is None: return
self.bpf_func = int(prog["bpf_func"]) aux = prog["aux"] self.btf = aux["btf"] linfo_idx = aux["linfo_idx"] self.nr_linfo = int(aux["nr_linfo"]) - linfo_idx if self.nr_linfo == 0: return
linfo_ptr = aux["linfo"] tpe = linfo_ptr.type.target().array(self.nr_linfo).pointer() self.linfo = (linfo_ptr + linfo_idx).cast(tpe).dereference() jited_linfo_ptr = aux["jited_linfo"] tpe = jited_linfo_ptr.type.target().array(self.nr_linfo).pointer() self.jited_linfo = (jited_linfo_ptr + linfo_idx).cast(tpe).dereference()
self.filenos = {}
def get_code_off(self): if self.pos >= self.nr_linfo: return -1 return self.jited_linfo[self.pos] - self.bpf_func
def advance(self): self.pos += 1
def get_fileno(self): file_name_off = int(self.linfo[self.pos]["file_name_off"]) fileno = self.filenos.get(file_name_off) if fileno is not None: return fileno, None file_name = btf_str_by_offset(self.btf, file_name_off) fileno = len(self.filenos) + 1 self.filenos[file_name_off] = fileno return fileno, file_name
def get_line_col(self): line_col = int(self.linfo[self.pos]["line_col"]) return bpf_line_info_line_num(line_col), \ bpf_line_info_line_col(line_col)
def generate_debug_obj(ksym, prog): name = get_ksym_name(ksym) # Avoid read_memory(); it throws bogus gdb.MemoryError in some contexts. start = ksym["start"] code = start.cast(gdb.lookup_type("unsigned char") .array(int(ksym["end"]) - int(start)) .pointer()).dereference().bytes linfo_iter = LInfoIter(prog)
result = tempfile.NamedTemporaryFile(suffix=".o", mode="wb") try: with tempfile.NamedTemporaryFile(suffix=".s", mode="w") as src: # ".loc" does not apply to ".byte"s, only to ".insn"s, but since # this needs to work for all architectures, the latter are not an # option. Ask the assembler to apply ".loc"s to labels as well, # and generate dummy labels after each ".loc". src.write(".loc_mark_labels 1\n")
src.write(".globl {}\n".format(name)) src.write(".type {},@function\n".format(name)) src.write("{}:\n".format(name)) for code_off, code_byte in enumerate(code): if linfo_iter.get_code_off() == code_off: fileno, file_name = linfo_iter.get_fileno() if file_name is not None: src.write(".file {} {}\n".format( fileno, json.dumps(file_name))) line, col = linfo_iter.get_line_col() src.write(".loc {} {} {}\n".format(fileno, line, col)) src.write("0:\n") linfo_iter.advance() src.write(".byte {}\n".format(code_byte)) src.write(".size {},{}\n".format(name, len(code))) src.flush()
try: subprocess.check_call(["as", "-c", src.name, "-o", result.name]) except FileNotFoundError: # "as" is not installed. result.close() return None return result except: result.close() raise