#!/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 osimport cmdimport argparseimport globfrom argparse import RawDescriptionHelpFormatterfrom argparse import RawTextHelpFormatterfrom 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 proceedingdebugfs_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.cclass 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 locationdef 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 devicedef 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 pathdef 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 devicedef 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 devicesdef 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 callingdef 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 callingdef 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 devicedef 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 devicedef 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 attributesdef 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 devicesdef 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 devicesdef 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 parserdef 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()