linux-yocto/tools/hv/vmbus_testing

#!/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/

/sys/kernel/debug/hyperv/UUID/

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()