vmbus_testing 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376
  1. #!/usr/bin/env python3
  2. # SPDX-License-Identifier: GPL-2.0
  3. #
  4. # Program to allow users to fuzz test Hyper-V drivers
  5. # by interfacing with Hyper-V debugfs attributes.
  6. # Current test methods available:
  7. # 1. delay testing
  8. #
  9. # Current file/directory structure of hyper-V debugfs:
  10. # /sys/kernel/debug/hyperv/UUID
  11. # /sys/kernel/debug/hyperv/UUID/<test-state filename>
  12. # /sys/kernel/debug/hyperv/UUID/<test-method sub-directory>
  13. #
  14. # author: Branden Bonaby <brandonbonaby94@gmail.com>
  15. import os
  16. import cmd
  17. import argparse
  18. import glob
  19. from argparse import RawDescriptionHelpFormatter
  20. from argparse import RawTextHelpFormatter
  21. from enum import Enum
  22. # Do not change unless, you change the debugfs attributes
  23. # in /drivers/hv/debugfs.c. All fuzz testing
  24. # attributes will start with "fuzz_test".
  25. # debugfs path for hyperv must exist before proceeding
  26. debugfs_hyperv_path = "/sys/kernel/debug/hyperv"
  27. if not os.path.isdir(debugfs_hyperv_path):
  28. print("{} doesn't exist/check permissions".format(debugfs_hyperv_path))
  29. exit(-1)
  30. class dev_state(Enum):
  31. off = 0
  32. on = 1
  33. # File names, that correspond to the files created in
  34. # /drivers/hv/debugfs.c
  35. class f_names(Enum):
  36. state_f = "fuzz_test_state"
  37. buff_f = "fuzz_test_buffer_interrupt_delay"
  38. mess_f = "fuzz_test_message_delay"
  39. # Both single_actions and all_actions are used
  40. # for error checking and to allow for some subparser
  41. # names to be abbreviated. Do not abbreviate the
  42. # test method names, as it will become less intuitive
  43. # as to what the user can do. If you do decide to
  44. # abbreviate the test method name, make sure the main
  45. # function reflects this change.
  46. all_actions = [
  47. "disable_all",
  48. "D",
  49. "enable_all",
  50. "view_all",
  51. "V"
  52. ]
  53. single_actions = [
  54. "disable_single",
  55. "d",
  56. "enable_single",
  57. "view_single",
  58. "v"
  59. ]
  60. def main():
  61. file_map = recursive_file_lookup(debugfs_hyperv_path, dict())
  62. args = parse_args()
  63. if (not args.action):
  64. print ("Error, no options selected...exiting")
  65. exit(-1)
  66. arg_set = { k for (k,v) in vars(args).items() if v and k != "action" }
  67. arg_set.add(args.action)
  68. path = args.path if "path" in arg_set else None
  69. if (path and path[-1] == "/"):
  70. path = path[:-1]
  71. validate_args_path(path, arg_set, file_map)
  72. if (path and "enable_single" in arg_set):
  73. state_path = locate_state(path, file_map)
  74. set_test_state(state_path, dev_state.on.value, args.quiet)
  75. # Use subparsers as the key for different actions
  76. if ("delay" in arg_set):
  77. validate_delay_values(args.delay_time)
  78. if (args.enable_all):
  79. set_delay_all_devices(file_map, args.delay_time,
  80. args.quiet)
  81. else:
  82. set_delay_values(path, file_map, args.delay_time,
  83. args.quiet)
  84. elif ("disable_all" in arg_set or "D" in arg_set):
  85. disable_all_testing(file_map)
  86. elif ("disable_single" in arg_set or "d" in arg_set):
  87. disable_testing_single_device(path, file_map)
  88. elif ("view_all" in arg_set or "V" in arg_set):
  89. get_all_devices_test_status(file_map)
  90. elif ("view_single" in arg_set or "v" in arg_set):
  91. get_device_test_values(path, file_map)
  92. # Get the state location
  93. def locate_state(device, file_map):
  94. return file_map[device][f_names.state_f.value]
  95. # Validate delay values to make sure they are acceptable to
  96. # enable delays on a device
  97. def validate_delay_values(delay):
  98. if (delay[0] == -1 and delay[1] == -1):
  99. print("\nError, At least 1 value must be greater than 0")
  100. exit(-1)
  101. for i in delay:
  102. if (i < -1 or i == 0 or i > 1000):
  103. print("\nError, Values must be equal to -1 "
  104. "or be > 0 and <= 1000")
  105. exit(-1)
  106. # Validate argument path
  107. def validate_args_path(path, arg_set, file_map):
  108. if (not path and any(element in arg_set for element in single_actions)):
  109. print("Error, path (-p) REQUIRED for the specified option. "
  110. "Use (-h) to check usage.")
  111. exit(-1)
  112. elif (path and any(item in arg_set for item in all_actions)):
  113. print("Error, path (-p) NOT REQUIRED for the specified option. "
  114. "Use (-h) to check usage." )
  115. exit(-1)
  116. elif (path not in file_map and any(item in arg_set
  117. for item in single_actions)):
  118. print("Error, path '{}' not a valid vmbus device".format(path))
  119. exit(-1)
  120. # display Testing status of single device
  121. def get_device_test_values(path, file_map):
  122. for name in file_map[path]:
  123. file_location = file_map[path][name]
  124. print( name + " = " + str(read_test_files(file_location)))
  125. # Create a map of the vmbus devices and their associated files
  126. # [key=device, value = [key = filename, value = file path]]
  127. def recursive_file_lookup(path, file_map):
  128. for f_path in glob.iglob(path + '**/*'):
  129. if (os.path.isfile(f_path)):
  130. if (f_path.rsplit("/",2)[0] == debugfs_hyperv_path):
  131. directory = f_path.rsplit("/",1)[0]
  132. else:
  133. directory = f_path.rsplit("/",2)[0]
  134. f_name = f_path.split("/")[-1]
  135. if (file_map.get(directory)):
  136. file_map[directory].update({f_name:f_path})
  137. else:
  138. file_map[directory] = {f_name:f_path}
  139. elif (os.path.isdir(f_path)):
  140. recursive_file_lookup(f_path,file_map)
  141. return file_map
  142. # display Testing state of devices
  143. def get_all_devices_test_status(file_map):
  144. for device in file_map:
  145. if (get_test_state(locate_state(device, file_map)) is 1):
  146. print("Testing = ON for: {}"
  147. .format(device.split("/")[5]))
  148. else:
  149. print("Testing = OFF for: {}"
  150. .format(device.split("/")[5]))
  151. # read the vmbus device files, path must be absolute path before calling
  152. def read_test_files(path):
  153. try:
  154. with open(path,"r") as f:
  155. file_value = f.readline().strip()
  156. return int(file_value)
  157. except IOError as e:
  158. errno, strerror = e.args
  159. print("I/O error({0}): {1} on file {2}"
  160. .format(errno, strerror, path))
  161. exit(-1)
  162. except ValueError:
  163. print ("Element to int conversion error in: \n{}".format(path))
  164. exit(-1)
  165. # writing to vmbus device files, path must be absolute path before calling
  166. def write_test_files(path, value):
  167. try:
  168. with open(path,"w") as f:
  169. f.write("{}".format(value))
  170. except IOError as e:
  171. errno, strerror = e.args
  172. print("I/O error({0}): {1} on file {2}"
  173. .format(errno, strerror, path))
  174. exit(-1)
  175. # set testing state of device
  176. def set_test_state(state_path, state_value, quiet):
  177. write_test_files(state_path, state_value)
  178. if (get_test_state(state_path) is 1):
  179. if (not quiet):
  180. print("Testing = ON for device: {}"
  181. .format(state_path.split("/")[5]))
  182. else:
  183. if (not quiet):
  184. print("Testing = OFF for device: {}"
  185. .format(state_path.split("/")[5]))
  186. # get testing state of device
  187. def get_test_state(state_path):
  188. #state == 1 - test = ON
  189. #state == 0 - test = OFF
  190. return read_test_files(state_path)
  191. # write 1 - 1000 microseconds, into a single device using the
  192. # fuzz_test_buffer_interrupt_delay and fuzz_test_message_delay
  193. # debugfs attributes
  194. def set_delay_values(device, file_map, delay_length, quiet):
  195. try:
  196. interrupt = file_map[device][f_names.buff_f.value]
  197. message = file_map[device][f_names.mess_f.value]
  198. # delay[0]- buffer interrupt delay, delay[1]- message delay
  199. if (delay_length[0] >= 0 and delay_length[0] <= 1000):
  200. write_test_files(interrupt, delay_length[0])
  201. if (delay_length[1] >= 0 and delay_length[1] <= 1000):
  202. write_test_files(message, delay_length[1])
  203. if (not quiet):
  204. print("Buffer delay testing = {} for: {}"
  205. .format(read_test_files(interrupt),
  206. interrupt.split("/")[5]))
  207. print("Message delay testing = {} for: {}"
  208. .format(read_test_files(message),
  209. message.split("/")[5]))
  210. except IOError as e:
  211. errno, strerror = e.args
  212. print("I/O error({0}): {1} on files {2}{3}"
  213. .format(errno, strerror, interrupt, message))
  214. exit(-1)
  215. # enabling delay testing on all devices
  216. def set_delay_all_devices(file_map, delay, quiet):
  217. for device in (file_map):
  218. set_test_state(locate_state(device, file_map),
  219. dev_state.on.value,
  220. quiet)
  221. set_delay_values(device, file_map, delay, quiet)
  222. # disable all testing on a SINGLE device.
  223. def disable_testing_single_device(device, file_map):
  224. for name in file_map[device]:
  225. file_location = file_map[device][name]
  226. write_test_files(file_location, dev_state.off.value)
  227. print("ALL testing now OFF for {}".format(device.split("/")[-1]))
  228. # disable all testing on ALL devices
  229. def disable_all_testing(file_map):
  230. for device in file_map:
  231. disable_testing_single_device(device, file_map)
  232. def parse_args():
  233. parser = argparse.ArgumentParser(prog = "vmbus_testing",usage ="\n"
  234. "%(prog)s [delay] [-h] [-e|-E] -t [-p]\n"
  235. "%(prog)s [view_all | V] [-h]\n"
  236. "%(prog)s [disable_all | D] [-h]\n"
  237. "%(prog)s [disable_single | d] [-h|-p]\n"
  238. "%(prog)s [view_single | v] [-h|-p]\n"
  239. "%(prog)s --version\n",
  240. description = "\nUse lsvmbus to get vmbus device type "
  241. "information.\n" "\nThe debugfs root path is "
  242. "/sys/kernel/debug/hyperv",
  243. formatter_class = RawDescriptionHelpFormatter)
  244. subparsers = parser.add_subparsers(dest = "action")
  245. parser.add_argument("--version", action = "version",
  246. version = '%(prog)s 0.1.0')
  247. parser.add_argument("-q","--quiet", action = "store_true",
  248. help = "silence none important test messages."
  249. " This will only work when enabling testing"
  250. " on a device.")
  251. # Use the path parser to hold the --path attribute so it can
  252. # be shared between subparsers. Also do the same for the state
  253. # parser, as all testing methods will use --enable_all and
  254. # enable_single.
  255. path_parser = argparse.ArgumentParser(add_help=False)
  256. path_parser.add_argument("-p","--path", metavar = "",
  257. help = "Debugfs path to a vmbus device. The path "
  258. "must be the absolute path to the device.")
  259. state_parser = argparse.ArgumentParser(add_help=False)
  260. state_group = state_parser.add_mutually_exclusive_group(required = True)
  261. state_group.add_argument("-E", "--enable_all", action = "store_const",
  262. const = "enable_all",
  263. help = "Enable the specified test type "
  264. "on ALL vmbus devices.")
  265. state_group.add_argument("-e", "--enable_single",
  266. action = "store_const",
  267. const = "enable_single",
  268. help = "Enable the specified test type on a "
  269. "SINGLE vmbus device.")
  270. parser_delay = subparsers.add_parser("delay",
  271. parents = [state_parser, path_parser],
  272. help = "Delay the ring buffer interrupt or the "
  273. "ring buffer message reads in microseconds.",
  274. prog = "vmbus_testing",
  275. usage = "%(prog)s [-h]\n"
  276. "%(prog)s -E -t [value] [value]\n"
  277. "%(prog)s -e -t [value] [value] -p",
  278. description = "Delay the ring buffer interrupt for "
  279. "vmbus devices, or delay the ring buffer message "
  280. "reads for vmbus devices (both in microseconds). This "
  281. "is only on the host to guest channel.")
  282. parser_delay.add_argument("-t", "--delay_time", metavar = "", nargs = 2,
  283. type = check_range, default =[0,0], required = (True),
  284. help = "Set [buffer] & [message] delay time. "
  285. "Value constraints: -1 == value "
  286. "or 0 < value <= 1000.\n"
  287. "Use -1 to keep the previous value for that delay "
  288. "type, or a value > 0 <= 1000 to change the delay "
  289. "time.")
  290. parser_dis_all = subparsers.add_parser("disable_all",
  291. aliases = ['D'], prog = "vmbus_testing",
  292. usage = "%(prog)s [disable_all | D] -h\n"
  293. "%(prog)s [disable_all | D]\n",
  294. help = "Disable ALL testing on ALL vmbus devices.",
  295. description = "Disable ALL testing on ALL vmbus "
  296. "devices.")
  297. parser_dis_single = subparsers.add_parser("disable_single",
  298. aliases = ['d'],
  299. parents = [path_parser], prog = "vmbus_testing",
  300. usage = "%(prog)s [disable_single | d] -h\n"
  301. "%(prog)s [disable_single | d] -p\n",
  302. help = "Disable ALL testing on a SINGLE vmbus device.",
  303. description = "Disable ALL testing on a SINGLE vmbus "
  304. "device.")
  305. parser_view_all = subparsers.add_parser("view_all", aliases = ['V'],
  306. help = "View the test state for ALL vmbus devices.",
  307. prog = "vmbus_testing",
  308. usage = "%(prog)s [view_all | V] -h\n"
  309. "%(prog)s [view_all | V]\n",
  310. description = "This shows the test state for ALL the "
  311. "vmbus devices.")
  312. parser_view_single = subparsers.add_parser("view_single",
  313. aliases = ['v'],parents = [path_parser],
  314. help = "View the test values for a SINGLE vmbus "
  315. "device.",
  316. description = "This shows the test values for a SINGLE "
  317. "vmbus device.", prog = "vmbus_testing",
  318. usage = "%(prog)s [view_single | v] -h\n"
  319. "%(prog)s [view_single | v] -p")
  320. return parser.parse_args()
  321. # value checking for range checking input in parser
  322. def check_range(arg1):
  323. try:
  324. val = int(arg1)
  325. except ValueError as err:
  326. raise argparse.ArgumentTypeError(str(err))
  327. if val < -1 or val > 1000:
  328. message = ("\n\nvalue must be -1 or 0 < value <= 1000. "
  329. "Value program received: {}\n").format(val)
  330. raise argparse.ArgumentTypeError(message)
  331. return val
  332. if __name__ == "__main__":
  333. main()