diff --git a/automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py b/automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py index 992d20e9c3..8e4654c115 100644 --- a/automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py +++ b/automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py @@ -27,6 +27,9 @@ import csv from requests import get import json +import requests.packages.urllib3 + +import modules.new_arguments2 SPLUNK_CONTAINER_APPS_DIR = "/opt/splunk/etc/apps" index_file_local_path = "indexes.conf.tar" @@ -38,25 +41,28 @@ datamodel_file_container_path = os.path.join(SPLUNK_CONTAINER_APPS_DIR, "Splunk_ MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING=2 -DEFAULT_CONTAINER_TAG="latest" -LOCAL_BASE_CONTAINER_NAME = "splunk_test_%d" -BASE_CONTAINER_WEB_PORT=8000 -BASE_CONTAINER_MANAGEMENT_PORT=8089 - - -DETECTION_TYPES = ['endpoint', 'cloud', 'network'] -DETECTION_MODES = ['new', 'all', 'selected'] def main(args): - - start_time = timer() - + requests.packages.urllib3.disable_warnings() + + start_datetime = datetime.now() + action, settings = modules.new_arguments2.parse(args) + if action == "configure": + #Done, nothing else to do + print("Configuration complete!") + sys.exit(0) + elif action != "run": + print("Unsupported action: [%s]"%(action), file=sys.stderr) + sys.exit(1) + print("time to run the test!") + sys.exit(0) + parser = argparse.ArgumentParser(description="CI Detection Testing") parser.add_argument("-b", "--branch", type=str, required=True, help="security content branch") @@ -92,9 +98,9 @@ def main(args): parser.add_argument("-split","--split_detections_then_stop", required=False, default=False, action='store_true', help="Should existing images be re-used, or should they be redownloaded?") - start_datetime = datetime.now() - import requests.packages.urllib3 - requests.packages.urllib3.disable_warnings() + + + args = parser.parse_args() branch = args.branch uuid_test = args.uuid diff --git a/automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py b/automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py index 4375b08e4a..460d9af287 100644 --- a/automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py +++ b/automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py @@ -1,12 +1,12 @@ import argparse import json -from typing import OrderedDict -import validate_args +from typing import OrderedDict, Union +import modules.validate_args as validate_args import sys DEFAULT_CONFIG_FILE = "test_config.json" -def configure_action(args) -> bool: +def configure_action(args) -> tuple[str, dict]: settings = OrderedDict() if args.input_config_file is None: settings, schema = validate_args.validate({}) @@ -58,16 +58,18 @@ def configure_action(args) -> bool: validated_new_settings, schema = validate_args.validate(new_config) if validated_new_settings == None: print("Error in the new settings!") - return False + sys.exit(1) + else: - print("New settings worked great. Writing results to: %s" % + print("New settings successful. Writing results to: %s" % (args.output_config_file.name)) args.output_config_file.write(json.dumps( validated_new_settings, sort_keys=True, indent=4)) - return True + + return ("configure", validated_new_settings) -def update_config_with_cli_arguments(args_dict:dict)->dict: +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']) @@ -86,19 +88,19 @@ def update_config_with_cli_arguments(args_dict:dict)->dict: print("Failure while processing updated settings from command line.\n\tQuitting...", file=sys.stderr) sys.exit(1) - return settings + return ("run", settings) -def run_action(args) -> bool: +def run_action(args) -> tuple[str,dict]: config = update_config_with_cli_arguments(args.__dict__) - return True + return config -def main(args): +def parse(args)->tuple[str,dict]: ''' try: with open(DEFAULT_CONFIG_FILE, 'r') as settings_file: @@ -158,12 +160,8 @@ def main(args): # Run the appropriate parser try: - if args.func(args): - print("Success!") - sys.exit(0) - else: - print("Fail") - sys.exit(1) + action, settings = args.func(args) + return action, settings except Exception as e: print("Unknown Error - [%s]" % (str(e))) sys.exit(1) @@ -251,4 +249,4 @@ def main(args): if __name__ == "__main__": - main(sys.argv[1:]) + parse(sys.argv[1:]) diff --git a/automated_detection_testing/ci/detection_testing_batch/modules/validate_args.py b/automated_detection_testing/ci/detection_testing_batch/modules/validate_args.py index 624ae90e5e..af57245555 100644 --- a/automated_detection_testing/ci/detection_testing_batch/modules/validate_args.py +++ b/automated_detection_testing/ci/detection_testing_batch/modules/validate_args.py @@ -4,7 +4,7 @@ import io import json import jsonschema import jsonschema.exceptions -import jsonschema_errorprinter +import modules.jsonschema_errorprinter as jsonschema_errorprinter import sys from typing import Union @@ -28,17 +28,17 @@ setup_schema = { "type": "boolean", "default": False }, - + "detections_list": { - "type":["array"], + "type": ["array"], "items": { - "type":"string" + "type": "string" }, - "default":[], + "default": [], }, - "detections_file":{ - "type": ["string","null"], + "detections_file": { + "type": ["string", "null"], "default": None }, @@ -57,11 +57,20 @@ setup_schema = { "type": "string" }, "local_path": { - "type": "string" + "type": ["string", "null"], + "default": None }, - }, - "default": [] - } + } + }, + "default": [ + { + "app_name": "SPLUNK_ES_CONTENT_UPDATE", + "app_number": 3449, + "app_version": "GENERATED", + 'local_path': None + } + ] + }, "mode": { @@ -214,53 +223,53 @@ setup_schema = { } -def validate_file(file:io.TextIOWrapper)->tuple[Union[dict, None], dict]: +def validate_file(file: io.TextIOWrapper) -> tuple[Union[dict, None], dict]: try: settings = json.loads(file.read()) return validate(settings) except Exception as e: raise(e) - - -def check_dependencies(settings: dict)->bool: - #Check complex mode dependencies +def check_dependencies(settings: dict) -> bool: + # Check complex mode dependencies error_free = True if settings['mode'] == 'selected': - #Make sure that exactly one of the following fields is populated + # Make sure that exactly one of the following fields is populated if settings['detections_file'] == None and settings['detections_list'] == []: - print("Error - mode was 'selected' but no detections_list or detections_file were supplied.",file=sys.stderr) + print("Error - mode was 'selected' but no detections_list or detections_file were supplied.", file=sys.stderr) error_free = False elif settings['detections_file'] != None and settings['detections_list'] != []: - print("Error - mode was 'selected' but detections_list and detections_file were supplied.",file=sys.stderr) + print("Error - mode was 'selected' but detections_list and detections_file were supplied.", file=sys.stderr) error_free = False - if settings['mode'] != 'selected'and settings['detections_file'] != None: - print("Error - mode was not 'selected' but detections_file was supplied.",file=sys.stderr) + if settings['mode'] != 'selected' and settings['detections_file'] != None: + print("Error - mode was not 'selected' but detections_file was supplied.", file=sys.stderr) error_free = False elif settings['mode'] != 'selected' and settings['detections_list'] != []: - print("Error - mode was not 'selected' but detections_list was supplied.",file=sys.stderr) + print("Error - mode was not 'selected' but detections_list was supplied.", file=sys.stderr) error_free = False - #Returns true if there are not errors + # Returns true if there are not errors return error_free + def validate(configuration: dict) -> tuple[Union[dict, None], dict]: #v = jsonschema.Draft201909Validator(argument_schema) try: validation_errors, validated_json = jsonschema_errorprinter.check_json( configuration, setup_schema) + no_complex_errors = check_dependencies(validated_json) if len(validation_errors) == 0 and no_complex_errors: print("Input configuration successfully validated!") return validated_json, setup_schema elif no_complex_errors == False: print("Failed due to error(s) listed above.", file=sys.stderr) - return None,setup_schema + return None, setup_schema else: print("[%d] failures detected during validation of the configuration!" % ( - len(validation_errors))) + len(validation_errors)),file=sys.stderr) for error in validation_errors: print(error, end="\n\n", file=sys.stderr) return None, setup_schema