| #!/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 <brandonbonaby94@gmail.com> | |
| 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)) == 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) == 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() |