forked from torvalds/linux
-
Notifications
You must be signed in to change notification settings - Fork 0
/
vmbus_testing
executable file
·376 lines (333 loc) · 16.1 KB
/
vmbus_testing
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
#
# Program to allow users to fuzz test Hyper-V drivers
# by interfacing with Hyper-V debugfs attributes.
# Current test methods available:
# 1. delay testing
#
# Current file/directory structure of hyper-V debugfs:
# /sys/kernel/debug/hyperv/UUID
# /sys/kernel/debug/hyperv/UUID/<test-state filename>
# /sys/kernel/debug/hyperv/UUID/<test-method sub-directory>
#
# author: Branden Bonaby <[email protected]>
import os
import cmd
import argparse
import glob
from argparse import RawDescriptionHelpFormatter
from argparse import RawTextHelpFormatter
from enum import Enum
# Do not change unless, you change the debugfs attributes
# in /drivers/hv/debugfs.c. All fuzz testing
# attributes will start with "fuzz_test".
# debugfs path for hyperv must exist before proceeding
debugfs_hyperv_path = "/sys/kernel/debug/hyperv"
if not os.path.isdir(debugfs_hyperv_path):
print("{} doesn't exist/check permissions".format(debugfs_hyperv_path))
exit(-1)
class dev_state(Enum):
off = 0
on = 1
# File names, that correspond to the files created in
# /drivers/hv/debugfs.c
class f_names(Enum):
state_f = "fuzz_test_state"
buff_f = "fuzz_test_buffer_interrupt_delay"
mess_f = "fuzz_test_message_delay"
# Both single_actions and all_actions are used
# for error checking and to allow for some subparser
# names to be abbreviated. Do not abbreviate the
# test method names, as it will become less intuitive
# as to what the user can do. If you do decide to
# abbreviate the test method name, make sure the main
# function reflects this change.
all_actions = [
"disable_all",
"D",
"enable_all",
"view_all",
"V"
]
single_actions = [
"disable_single",
"d",
"enable_single",
"view_single",
"v"
]
def main():
file_map = recursive_file_lookup(debugfs_hyperv_path, dict())
args = parse_args()
if (not args.action):
print ("Error, no options selected...exiting")
exit(-1)
arg_set = { k for (k,v) in vars(args).items() if v and k != "action" }
arg_set.add(args.action)
path = args.path if "path" in arg_set else None
if (path and path[-1] == "/"):
path = path[:-1]
validate_args_path(path, arg_set, file_map)
if (path and "enable_single" in arg_set):
state_path = locate_state(path, file_map)
set_test_state(state_path, dev_state.on.value, args.quiet)
# Use subparsers as the key for different actions
if ("delay" in arg_set):
validate_delay_values(args.delay_time)
if (args.enable_all):
set_delay_all_devices(file_map, args.delay_time,
args.quiet)
else:
set_delay_values(path, file_map, args.delay_time,
args.quiet)
elif ("disable_all" in arg_set or "D" in arg_set):
disable_all_testing(file_map)
elif ("disable_single" in arg_set or "d" in arg_set):
disable_testing_single_device(path, file_map)
elif ("view_all" in arg_set or "V" in arg_set):
get_all_devices_test_status(file_map)
elif ("view_single" in arg_set or "v" in arg_set):
get_device_test_values(path, file_map)
# Get the state location
def locate_state(device, file_map):
return file_map[device][f_names.state_f.value]
# Validate delay values to make sure they are acceptable to
# enable delays on a device
def validate_delay_values(delay):
if (delay[0] == -1 and delay[1] == -1):
print("\nError, At least 1 value must be greater than 0")
exit(-1)
for i in delay:
if (i < -1 or i == 0 or i > 1000):
print("\nError, Values must be equal to -1 "
"or be > 0 and <= 1000")
exit(-1)
# Validate argument path
def validate_args_path(path, arg_set, file_map):
if (not path and any(element in arg_set for element in single_actions)):
print("Error, path (-p) REQUIRED for the specified option. "
"Use (-h) to check usage.")
exit(-1)
elif (path and any(item in arg_set for item in all_actions)):
print("Error, path (-p) NOT REQUIRED for the specified option. "
"Use (-h) to check usage." )
exit(-1)
elif (path not in file_map and any(item in arg_set
for item in single_actions)):
print("Error, path '{}' not a valid vmbus device".format(path))
exit(-1)
# display Testing status of single device
def get_device_test_values(path, file_map):
for name in file_map[path]:
file_location = file_map[path][name]
print( name + " = " + str(read_test_files(file_location)))
# Create a map of the vmbus devices and their associated files
# [key=device, value = [key = filename, value = file path]]
def recursive_file_lookup(path, file_map):
for f_path in glob.iglob(path + '**/*'):
if (os.path.isfile(f_path)):
if (f_path.rsplit("/",2)[0] == debugfs_hyperv_path):
directory = f_path.rsplit("/",1)[0]
else:
directory = f_path.rsplit("/",2)[0]
f_name = f_path.split("/")[-1]
if (file_map.get(directory)):
file_map[directory].update({f_name:f_path})
else:
file_map[directory] = {f_name:f_path}
elif (os.path.isdir(f_path)):
recursive_file_lookup(f_path,file_map)
return file_map
# display Testing state of devices
def get_all_devices_test_status(file_map):
for device in file_map:
if (get_test_state(locate_state(device, file_map)) is 1):
print("Testing = ON for: {}"
.format(device.split("/")[5]))
else:
print("Testing = OFF for: {}"
.format(device.split("/")[5]))
# read the vmbus device files, path must be absolute path before calling
def read_test_files(path):
try:
with open(path,"r") as f:
file_value = f.readline().strip()
return int(file_value)
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on file {2}"
.format(errno, strerror, path))
exit(-1)
except ValueError:
print ("Element to int conversion error in: \n{}".format(path))
exit(-1)
# writing to vmbus device files, path must be absolute path before calling
def write_test_files(path, value):
try:
with open(path,"w") as f:
f.write("{}".format(value))
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on file {2}"
.format(errno, strerror, path))
exit(-1)
# set testing state of device
def set_test_state(state_path, state_value, quiet):
write_test_files(state_path, state_value)
if (get_test_state(state_path) is 1):
if (not quiet):
print("Testing = ON for device: {}"
.format(state_path.split("/")[5]))
else:
if (not quiet):
print("Testing = OFF for device: {}"
.format(state_path.split("/")[5]))
# get testing state of device
def get_test_state(state_path):
#state == 1 - test = ON
#state == 0 - test = OFF
return read_test_files(state_path)
# write 1 - 1000 microseconds, into a single device using the
# fuzz_test_buffer_interrupt_delay and fuzz_test_message_delay
# debugfs attributes
def set_delay_values(device, file_map, delay_length, quiet):
try:
interrupt = file_map[device][f_names.buff_f.value]
message = file_map[device][f_names.mess_f.value]
# delay[0]- buffer interrupt delay, delay[1]- message delay
if (delay_length[0] >= 0 and delay_length[0] <= 1000):
write_test_files(interrupt, delay_length[0])
if (delay_length[1] >= 0 and delay_length[1] <= 1000):
write_test_files(message, delay_length[1])
if (not quiet):
print("Buffer delay testing = {} for: {}"
.format(read_test_files(interrupt),
interrupt.split("/")[5]))
print("Message delay testing = {} for: {}"
.format(read_test_files(message),
message.split("/")[5]))
except IOError as e:
errno, strerror = e.args
print("I/O error({0}): {1} on files {2}{3}"
.format(errno, strerror, interrupt, message))
exit(-1)
# enabling delay testing on all devices
def set_delay_all_devices(file_map, delay, quiet):
for device in (file_map):
set_test_state(locate_state(device, file_map),
dev_state.on.value,
quiet)
set_delay_values(device, file_map, delay, quiet)
# disable all testing on a SINGLE device.
def disable_testing_single_device(device, file_map):
for name in file_map[device]:
file_location = file_map[device][name]
write_test_files(file_location, dev_state.off.value)
print("ALL testing now OFF for {}".format(device.split("/")[-1]))
# disable all testing on ALL devices
def disable_all_testing(file_map):
for device in file_map:
disable_testing_single_device(device, file_map)
def parse_args():
parser = argparse.ArgumentParser(prog = "vmbus_testing",usage ="\n"
"%(prog)s [delay] [-h] [-e|-E] -t [-p]\n"
"%(prog)s [view_all | V] [-h]\n"
"%(prog)s [disable_all | D] [-h]\n"
"%(prog)s [disable_single | d] [-h|-p]\n"
"%(prog)s [view_single | v] [-h|-p]\n"
"%(prog)s --version\n",
description = "\nUse lsvmbus to get vmbus device type "
"information.\n" "\nThe debugfs root path is "
"/sys/kernel/debug/hyperv",
formatter_class = RawDescriptionHelpFormatter)
subparsers = parser.add_subparsers(dest = "action")
parser.add_argument("--version", action = "version",
version = '%(prog)s 0.1.0')
parser.add_argument("-q","--quiet", action = "store_true",
help = "silence none important test messages."
" This will only work when enabling testing"
" on a device.")
# Use the path parser to hold the --path attribute so it can
# be shared between subparsers. Also do the same for the state
# parser, as all testing methods will use --enable_all and
# enable_single.
path_parser = argparse.ArgumentParser(add_help=False)
path_parser.add_argument("-p","--path", metavar = "",
help = "Debugfs path to a vmbus device. The path "
"must be the absolute path to the device.")
state_parser = argparse.ArgumentParser(add_help=False)
state_group = state_parser.add_mutually_exclusive_group(required = True)
state_group.add_argument("-E", "--enable_all", action = "store_const",
const = "enable_all",
help = "Enable the specified test type "
"on ALL vmbus devices.")
state_group.add_argument("-e", "--enable_single",
action = "store_const",
const = "enable_single",
help = "Enable the specified test type on a "
"SINGLE vmbus device.")
parser_delay = subparsers.add_parser("delay",
parents = [state_parser, path_parser],
help = "Delay the ring buffer interrupt or the "
"ring buffer message reads in microseconds.",
prog = "vmbus_testing",
usage = "%(prog)s [-h]\n"
"%(prog)s -E -t [value] [value]\n"
"%(prog)s -e -t [value] [value] -p",
description = "Delay the ring buffer interrupt for "
"vmbus devices, or delay the ring buffer message "
"reads for vmbus devices (both in microseconds). This "
"is only on the host to guest channel.")
parser_delay.add_argument("-t", "--delay_time", metavar = "", nargs = 2,
type = check_range, default =[0,0], required = (True),
help = "Set [buffer] & [message] delay time. "
"Value constraints: -1 == value "
"or 0 < value <= 1000.\n"
"Use -1 to keep the previous value for that delay "
"type, or a value > 0 <= 1000 to change the delay "
"time.")
parser_dis_all = subparsers.add_parser("disable_all",
aliases = ['D'], prog = "vmbus_testing",
usage = "%(prog)s [disable_all | D] -h\n"
"%(prog)s [disable_all | D]\n",
help = "Disable ALL testing on ALL vmbus devices.",
description = "Disable ALL testing on ALL vmbus "
"devices.")
parser_dis_single = subparsers.add_parser("disable_single",
aliases = ['d'],
parents = [path_parser], prog = "vmbus_testing",
usage = "%(prog)s [disable_single | d] -h\n"
"%(prog)s [disable_single | d] -p\n",
help = "Disable ALL testing on a SINGLE vmbus device.",
description = "Disable ALL testing on a SINGLE vmbus "
"device.")
parser_view_all = subparsers.add_parser("view_all", aliases = ['V'],
help = "View the test state for ALL vmbus devices.",
prog = "vmbus_testing",
usage = "%(prog)s [view_all | V] -h\n"
"%(prog)s [view_all | V]\n",
description = "This shows the test state for ALL the "
"vmbus devices.")
parser_view_single = subparsers.add_parser("view_single",
aliases = ['v'],parents = [path_parser],
help = "View the test values for a SINGLE vmbus "
"device.",
description = "This shows the test values for a SINGLE "
"vmbus device.", prog = "vmbus_testing",
usage = "%(prog)s [view_single | v] -h\n"
"%(prog)s [view_single | v] -p")
return parser.parse_args()
# value checking for range checking input in parser
def check_range(arg1):
try:
val = int(arg1)
except ValueError as err:
raise argparse.ArgumentTypeError(str(err))
if val < -1 or val > 1000:
message = ("\n\nvalue must be -1 or 0 < value <= 1000. "
"Value program received: {}\n").format(val)
raise argparse.ArgumentTypeError(message)
return val
if __name__ == "__main__":
main()