mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
d1vious
235 lines
11 KiB
Python
235 lines
11 KiB
Python
import argparse
|
|
import json
|
|
from typing import OrderedDict, Union
|
|
from modules import validate_args
|
|
import sys
|
|
|
|
DEFAULT_CONFIG_FILE = "test_config.json"
|
|
|
|
|
|
def configure_action(args) -> tuple[str, dict]:
|
|
settings = OrderedDict()
|
|
if args.input_config_file is None:
|
|
settings, schema = validate_args.validate({})
|
|
else:
|
|
settings, schema = validate_args.validate_file(args.input_config_file)
|
|
|
|
if settings is None:
|
|
print("Failure while processing settings\n\tQuitting...", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
new_config = {}
|
|
for arg in settings:
|
|
default = settings[arg]
|
|
default_string = str(default).replace("'", '"')
|
|
|
|
if 'enum' in schema['properties'][arg]:
|
|
choice = input("%s [default: %s | choices: {%s}]: " % (
|
|
arg, default_string, ','.join(schema['properties'][arg]['enum'])))
|
|
else:
|
|
choice = input("%s [default: %s]: " % (arg, default_string))
|
|
choice = choice.strip()
|
|
if len(choice) == 0:
|
|
print("\tNothing entered, using default:")
|
|
new_config[arg] = default
|
|
formatted_print = default
|
|
else:
|
|
if choice.lower() in ["true", "false"] and schema['properties'][arg]['type'] == "boolean":
|
|
new_config[arg] = json.loads(choice.lower())
|
|
formatted_print = choice.lower()
|
|
else:
|
|
|
|
if choice in ['true', 'false'] or (choice.isdigit() and schema['properties'][arg]['type'] != "integer"):
|
|
choice = '"' + choice + '"'
|
|
# replace all single quotes with doubles quotes to make valid json
|
|
elif "'" in choice:
|
|
print('''Found %d single quotes (') in input... we will convert these to double quotes (") to ensure valida json.''' % (
|
|
choice.count("'")))
|
|
choice = choice.replace("'", '"')
|
|
elif '"' in choice:
|
|
# Do nothing
|
|
pass
|
|
elif choice.isdigit():
|
|
pass
|
|
else:
|
|
choice = '"' + choice + '"'
|
|
|
|
new_config[arg] = json.loads(choice)
|
|
formatted_print = choice
|
|
# We print out choice instead of new_config[arg] because the json.loads() messes up the quotation marks again
|
|
print("\t{0}\n".format(formatted_print))
|
|
|
|
# Now parse the new config and make sure it's good
|
|
validated_new_settings, schema = validate_args.validate_and_write(
|
|
new_config, args.output_config_file, skip_password_accessibility_check=False)
|
|
if validated_new_settings == None:
|
|
print("Could not update settings.\n\tQuitting...", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
return ("configure", validated_new_settings)
|
|
|
|
|
|
def update_config_with_cli_arguments(args_dict: dict) -> tuple[str, dict]:
|
|
# First load the config file
|
|
|
|
settings, _ = validate_args.validate_file(args_dict['config_file'])
|
|
if settings is None:
|
|
print("Failure while processing settings in [%s].\n\tQuitting..." % (
|
|
args_dict['config_file'].name), file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Then update it with the values that were passed as command line arguments
|
|
for key, value in args_dict.items():
|
|
if key in settings:
|
|
settings[key] = value
|
|
|
|
# Validate again to make sure we didn't break anything
|
|
settings, _ = validate_args.validate(settings,skip_password_accessibility_check=False)
|
|
if settings is None:
|
|
print("Failure while processing updated settings from command line.\n\tQuitting...", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
return ("run", settings)
|
|
|
|
|
|
def run_action(args) -> tuple[str, dict]:
|
|
|
|
config = update_config_with_cli_arguments(args.__dict__)
|
|
|
|
return config
|
|
|
|
|
|
def parse(args) -> tuple[str, dict]:
|
|
'''
|
|
try:
|
|
with open(DEFAULT_CONFIG_FILE, 'r') as settings_file:
|
|
default_settings = json.load(settings_file)
|
|
except Exception as e:
|
|
print("Error loading settings file %s: %s"%(DEFAULT_CONFIG_FILE, str(e)), file=sys.stderr)
|
|
sys.exit(1)
|
|
'''
|
|
|
|
import os
|
|
# if there is no default config file, then generate one
|
|
if not os.path.exists(DEFAULT_CONFIG_FILE):
|
|
print("No default configuration file [%s] found. Creating one..." % (
|
|
DEFAULT_CONFIG_FILE))
|
|
with open(DEFAULT_CONFIG_FILE, 'w') as cfg:
|
|
validate_args.validate_and_write({}, cfg, skip_password_accessibility_check=True)
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Use 'SOME_PROGRAM_NAME_STRING --help' to get help with the arguments")
|
|
parser.set_defaults(func=lambda _: parser.print_help())
|
|
|
|
actions_parser = parser.add_subparsers(title="Action")
|
|
|
|
# Configure parser
|
|
configure_parser = actions_parser.add_parser(
|
|
"configure", help="Configure a test run")
|
|
configure_parser.set_defaults(func=configure_action)
|
|
configure_parser.add_argument('-i', '--input_config_file', required=False,
|
|
type=argparse.FileType('r'), help="The config file to base the configuration off of.")
|
|
configure_parser.add_argument('-o', '--output_config_file', required=False, default=DEFAULT_CONFIG_FILE,
|
|
type=argparse.FileType('w'), help="The config file to write the configuration off of.")
|
|
|
|
# Run parser
|
|
run_parser = actions_parser.add_parser(
|
|
"run", help="Run a test")
|
|
run_parser.set_defaults(func=run_action)
|
|
run_parser.add_argument('-c', '--config_file', required=False,
|
|
type=argparse.FileType('r'),
|
|
default=DEFAULT_CONFIG_FILE,
|
|
help="The config file for the test. Note that this file "
|
|
"cannot be changed (except for credentials that can be "
|
|
"entered on the command line).")
|
|
|
|
run_parser.add_argument('-user', '--splunkbase_username', required=False, type=str,
|
|
help="Username for login to splunkbase. This is required "
|
|
"if downloading packages from Splunkbase. While this can "
|
|
"be stored in the config file, it is strongly recommended "
|
|
"to enter it at runtime.")
|
|
|
|
run_parser.add_argument('-b', '--branch', required=False, type=str,
|
|
help="The branch to run the tests on.")
|
|
|
|
run_parser.add_argument('-hash', '--commit_hash', required=False, type=str,
|
|
help="The hash to run the tests on.")
|
|
|
|
run_parser.add_argument('-pr', '--pr_number', required=False, type=int,
|
|
help="The Pull request to run the tests on.")
|
|
|
|
run_parser.add_argument('-m', '--mode', required=False, type=str,
|
|
help="The mode all, changes, or selected for the testing.")
|
|
|
|
run_parser.add_argument('-pass', '--splunkbase_password', required=False, type=str,
|
|
help="Password for login to splunkbase. This is required if "
|
|
"downloading packages from Splunkbase. While this can be "
|
|
"stored in the config file, it is strongly recommended "
|
|
"to enter it at runtime.")
|
|
|
|
run_parser.add_argument('-splunkpass', '--splunk_app_password', required=False, type=str,
|
|
help="Password for login to the splunk app. If you don't "
|
|
"provide one here or in the config, it will be generated "
|
|
"automatically for you.")
|
|
|
|
run_parser.add_argument("-show_pass", "--show_splunk_app_password", required=False,
|
|
action="store_true",
|
|
help="The password to login to the Splunk Server. If the config "
|
|
"file is set to true, it will override the default False for this. True "
|
|
"will override the default value in the config file.")
|
|
|
|
run_parser.add_argument("-mock", "--mock", required=False,
|
|
action="store_true",
|
|
help="Split into multiple configs, don't actually run the tests. If the config "
|
|
"file is set to true, it will override the default False for this. True "
|
|
"will override the default value in the config file.")
|
|
|
|
run_parser.add_argument("-n", "--num_containers", required=False, type=int,
|
|
help="The number of Splunk containers to run or mock")
|
|
|
|
run_parser.add_argument("-nif", "--no_interactive_failure", required=False,
|
|
action="store_true",
|
|
help="After a detection fails, pause and allow the user to log into "\
|
|
"the Splunk server to interactively debug the failure. Wait for the user "\
|
|
"to hit enter before removing the test data and moving on to the next test.")
|
|
|
|
run_parser.add_argument("-i", "--interactive", required=False,
|
|
action="store_true",
|
|
help="After a detection runs, pause and allow the user to log into "\
|
|
"the Splunk server to debug the detection. Wait for the user "\
|
|
"to hit enter before removing the test data and moving on to the next test.")
|
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
# Run the appropriate parser
|
|
try:
|
|
# If one of these arguments is not passed on the command line, don't overwrite its config
|
|
# file value with None - keep the config file value
|
|
keys = list(args.__dict__.keys())
|
|
for key in keys:
|
|
|
|
# We have to do the check separately because booleans using the --store_true
|
|
# action have an implict default=False value, even if we don't set it. We cannot
|
|
# set their value to something else, like None
|
|
|
|
# Don't overwite booleans
|
|
if args.__dict__[key] is False and key in ["show_splunk_app_password", "mock", "no_interactive_failure", "interactive"]:
|
|
del args.__dict__[key]
|
|
# Don't overwrite other values
|
|
elif args.__dict__[key] is None and key in ["splunkbase_username", "branch", "commit_hash",
|
|
"pr_number", "mode", "splunkbase_password",
|
|
"num_containers"]:
|
|
del args.__dict__[key]
|
|
|
|
action, settings = args.func(args)
|
|
|
|
|
|
return action, settings
|
|
except Exception as e:
|
|
print("Unknown Error Validating Json Configuration - [%s]" % (str(e)))
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
parse(sys.argv[1:])
|