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