1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
#!/usr/bin/env python3
# SPDX-License-Identifier: BSD-3-Clause
#
# callwalk.py - walk the Open-IOV callgraph index and print callee trees /
# caller trees around a function. This is the quick way to go from an export
# wrapper to the internal implementation chain without opening Ghidra.
#
# Usage:
# callwalk.py <callgraph.tsv> callees <hex_file_vaddr> [depth] [max_per_node]
# callwalk.py <callgraph.tsv> callers <hex_file_vaddr> [depth] [max_per_node]
# callwalk.py <callgraph.tsv> chain <hex_from> <hex_to> # shortest path
#
# Addresses are FILE virtual addresses (libcuda+0x...); the callgraph columns
# `caller_file`/`callee_file` are used, so Ghidra image-base offsets do not
# matter. `names.tsv` (file_vaddr<TAB>name) is picked up next to the callgraph
# when present, and used to label nodes.
#
# SPDX-License-Identifier: BSD-3-Clause
import sys
from collections import defaultdict, deque
def load_graph(path):
callees = defaultdict(list)
callers = defaultdict(list)
with open(path) as f:
header = f.readline()
for line in f:
parts = line.rstrip("\n").split("\t")
if len(parts) < 4:
continue
cf, kf = int(parts[2], 16), int(parts[3], 16)
callees[cf].append(kf)
callers[kf].append(cf)
return callees, callers
def load_names(path):
names = {}
try:
with open(path) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
parts = line.split("\t")
if len(parts) >= 2:
try:
names[int(parts[0], 16)] = parts[1]
except ValueError:
pass
except FileNotFoundError:
pass
return names
def label(a, names):
return f"0x{a:08x}({names[a]})" if a in names else f"0x{a:08x}"
def walk(adj, start, depth, limit):
seen = {start}
frontier = [start]
for d in range(depth):
nxt = []
for node in frontier:
kids = sorted(set(adj.get(node, [])))
if not kids:
continue
print(" " * (d + 1) + " | ".join(label(k, NAMES) for k in kids[:limit])
+ (" ..." if len(kids) > limit else ""))
for k in kids:
if k not in seen:
seen.add(k)
nxt.append(k)
frontier = nxt
if not frontier:
break
def shortest_path(callees, start, goal):
if start == goal:
return [start]
prev = {start: None}
q = deque([start])
while q:
node = q.popleft()
for k in callees.get(node, []):
if k in prev:
continue
prev[k] = node
if k == goal:
path = [goal]
while prev[path[-1]] is not None:
path.append(prev[path[-1]])
return list(reversed(path))
q.append(k)
return None
def main():
if len(sys.argv) < 4:
print(__doc__)
sys.exit(2)
graph = sys.argv[1]
mode = sys.argv[2]
addr = int(sys.argv[3], 16)
depth = int(sys.argv[4]) if len(sys.argv) > 4 else 4
limit = int(sys.argv[5]) if len(sys.argv) > 5 else 8
global NAMES
callees, callers = load_graph(graph)
NAMES = load_names(graph.rsplit("/", 1)[0] + "/names.tsv")
if mode == "chain":
goal = int(sys.argv[4], 16)
path = shortest_path(callees, addr, goal)
if not path:
print("no direct path found")
else:
for a in path:
print(label(a, NAMES))
return
print(f"{mode} of {label(addr, NAMES)}:")
walk(callees if mode == "callees" else callers, addr, depth, limit)
NAMES = {}
if __name__ == "__main__":
main()
|
