From dab746bb0c7ebd92b41af4628087382bd8f76cec Mon Sep 17 00:00:00 2001 From: pyth0n1c <87383215+pyth0n1c@users.noreply.github.com> Date: Tue, 23 Nov 2021 14:30:54 -0800 Subject: [PATCH] Save out a json file showing the config, to include the command line arguments with credentials removed, whenever a test is actually run. This makes it trivial to reproduce the test run on another machine or again on the same machine. --- .../detection_testing_execution.py | 23 ++++++++++--------- .../modules/container_manager.py | 2 +- .../modules/github_service.py | 11 +++++---- .../modules/new_arguments2.py | 21 ++++++++++++++++- .../modules/splunk_container.py | 20 ++++++++-------- 5 files changed, 49 insertions(+), 28 deletions(-) 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 2a25ce999f..b3ef52ac68 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 @@ -48,10 +48,10 @@ def copy_local_apps_to_directory(apps: dict[str,dict], target_directory)->None: source_path = os.path.abspath(os.path.expanduser(item['local_path'])) base_name = os.path.basename(source_path) dest_path = os.path.join(target_directory, base_name) + try: shutil.copy(source_path, dest_path) item['local_path'] = dest_path - print("Copied %s to apps"%(base_name)) except shutil.SameFileError as e: # Same file, not a real error. The copy just doesn't happen print("err:%s"%(str(e))) @@ -153,6 +153,7 @@ def generate_escu_app(persist_security_content: bool = False) -> str: response.raise_for_status() with open(SPLUNK_PACKAGING_TOOLKIT_FILENAME, 'wb') as slim_file: slim_file.write(response.content) + print("Done") except Exception as e: print("Error downloading the Splunk Packaging Toolkit: [%s].\n\tQuitting..." % (str(e)), file=sys.stderr) @@ -186,6 +187,11 @@ def generate_escu_app(persist_security_content: bool = False) -> str: def main(args: list[str]): + try: + docker.client.from_env() + except Exception as e: + print("Error, failed to get docker client. Is Docker Running?\n\t%s"%(str(e))) + requests.packages.urllib3.disable_warnings() start_datetime = datetime.now() @@ -197,8 +203,8 @@ def main(args: list[str]): elif action != "run": print("Unsupported action: [%s]" % (action), file=sys.stderr) sys.exit(1) - - + + ''' parser = argparse.ArgumentParser(description="CI Detection Testing") parser.add_argument("-b", "--branch", type=str, required=True, help="security content branch") @@ -274,13 +280,8 @@ def main(args: list[str]): settings['detections_list'], settings['detections_file']) - -# if len(all_test_files) == 0: -# print("No files were found to be tested. While this could be due to an error, "\ -# "this could be correct if there were no changes to detections. We will "\ -# "exit with success.\n\tQuitting...") -# sys.exit(0) - + + local_volume_absolute_path = os.path.abspath( os.path.join(os.getcwd(), "apps")) try: @@ -444,7 +445,7 @@ def main(args: list[str]): interactive_failure=settings['interactive_failure']) - print(cm.containers[0].environment) + cm.run_test() ''' diff --git a/automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py b/automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py index a7136ea937..4e45a9010d 100644 --- a/automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py +++ b/automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py @@ -103,7 +103,7 @@ class ContainerManager: self.baseline['TEST_FINISH_TIME'] = str(stop_time) duration = stop_time - self.start_time - self.baseline['TEST_DURATION'] = duration - datetime.timedelta(microseconds=duration.microseconds) + self.baseline['TEST_DURATION'] = str(duration - datetime.timedelta(microseconds=duration.microseconds)) self.synchronization_object.finish(self.baseline) diff --git a/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py b/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py index 5450279021..f1e88a3dae 100644 --- a/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py +++ b/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py @@ -50,6 +50,7 @@ class GithubService: summary_file: str = None) -> list[str]: pruned_tests = [] csvlines = [] + for detection in detections_to_prune: if os.path.basename(detection).startswith("ssa") and exclude_ssa: continue @@ -61,8 +62,8 @@ class GithubService: test_filepath_without_security_content = str( pathlib.Path(*pathlib.Path(test_filepath).parts[1:])) # If no types are provided, then we will get everything - if 'type' in description and (description['type'] in types_to_test or len( types_to_test) == 0): - # print(description['type']) + if 'type' in description and (description['type'] in types_to_test or len(types_to_test) == 0): + if not os.path.exists(test_filepath): print("Detection [%s] references [%s], but it does not exist" % ( detection, test_filepath)) @@ -124,7 +125,7 @@ class GithubService: (detections_list, detections_file), file=sys.stderr) sys.exit(1) elif detections_list is not None: - tests = self.get_selected_test_files(detections_list, folders, types) + tests = self.get_selected_test_files(detections_list, types) elif detections_file is not None: try: with open(detections_file,'r') as f: @@ -156,7 +157,7 @@ class GithubService: "Anomaly", "Hunting", "TTP"], previously_successful_tests: list[str] = []) -> list[str]: - return self.prune_detections(detection_file_list, types_to_test, previously_successful_tests) + return self.prune_detections(detection_file_list, types_to_test, previously_successful_tests) def get_all_tests_and_detections(self, folders: list[str] = [ @@ -170,7 +171,7 @@ class GithubService: os.path.join("security_content/detections", folder), "*.yml")) # Prune this down to only the subset of detections we can test - return self.prune_detections(detections, types_to_test, previously_successful_tests) + return self.prune_detections(detections, types_to_test, previously_successful_tests) def get_all_files_in_folder(self, foldername: str, extension: str) -> list[str]: filenames = glob.glob(os.path.join(foldername, extension)) 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 978fcdafd8..2ec757b9a9 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,4 +1,6 @@ import argparse +import copy +import datetime import json from typing import OrderedDict, Union import modules.validate_args as validate_args @@ -86,6 +88,16 @@ def update_config_with_cli_arguments(args_dict:dict)->tuple[str, dict]: print("Failure while processing updated settings from command line.\n\tQuitting...", file=sys.stderr) sys.exit(1) + now = datetime.datetime.now() + configname = now.strftime('%Y-%m-%dT%H:%M:%S.%f%z') + '-test-run.json' + with open(configname,'w') as test_config: + settings_with_creds_stripped = copy.deepcopy(settings) + #strip out credentials + settings_with_creds_stripped['splunkbase_password'] = None + settings_with_creds_stripped['splunkbase_username'] = None + settings_with_creds_stripped['container_password'] = None + validate_args.validate_and_write(settings_with_creds_stripped, test_config) + return ("run", settings) @@ -141,6 +153,13 @@ def parse(args)->tuple[str,dict]: "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('-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 " @@ -158,7 +177,7 @@ def parse(args)->tuple[str,dict]: "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("-m", "--mock", required=False, + 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 "\ diff --git a/automated_detection_testing/ci/detection_testing_batch/modules/splunk_container.py b/automated_detection_testing/ci/detection_testing_batch/modules/splunk_container.py index 1318eda110..1c9f3ff0f0 100644 --- a/automated_detection_testing/ci/detection_testing_batch/modules/splunk_container.py +++ b/automated_detection_testing/ci/detection_testing_batch/modules/splunk_container.py @@ -59,7 +59,7 @@ class SplunkContainer: self.management_port = management_port_tuple[1] self.container = self.make_container() - self.thread = threading.Thread(target=self.run_container) + self.thread = threading.Thread(target=self.run_container, ) self.container_start_time = 0 self.test_start_time = 0 @@ -221,7 +221,7 @@ class SplunkContainer: if self.container_start_time == -1: total_time_string = "NOT STARTED" else: - total_time_rounded = datetime.timedelta( + total_time_rounded = datetime.timedelta( seconds = round(current_time - self.container_start_time)) total_time_string = str(total_time_rounded) @@ -229,7 +229,7 @@ class SplunkContainer: if self.test_start_time == -1 or self.container_start_time == -1: setup_time_string = "NOT SET UP" else: - setup_secounds_rounded = datetime.timedelta( + setup_secounds_rounded = datetime.timedelta(seconds = round(self.test_start_time - self.container_start_time)) setup_time_string = str(setup_secounds_rounded) @@ -237,7 +237,7 @@ class SplunkContainer: if self.test_start_time == -1 or self.num_tests_completed == 0: testing_time_string = "NO TESTS COMPLETED" else: - testing_seconds_rounded = datetime.timedelta( + testing_seconds_rounded = datetime.timedelta( seconds = round(current_time - self.test_start_time)) # Get the approximate time per test. This is a clunky way to get rid of decimal @@ -246,13 +246,13 @@ class SplunkContainer: timedelta_per_test_rounded = timedelta_per_test - \ datetime.timedelta( microseconds=timedelta_per_test.microseconds) + + testing_time_string = "%s per test (%d tests)"%(timedelta_per_test_rounded, self.num_tests_completed) - testing_time_string = "%s per test (%d tests)"%(timedelta_per_test_rounded, str(testing_seconds_rounded)) - - summary_str = "[%s] Summary\n\t"\ - "Total Time :"\ - "Container Start Time:"\ - "Test Execution Time :" %(total_time_string, setup_time_string, testing_time_string) + summary_str = "Summary\n\t"\ + "Total Time : [%s]\n\t"\ + "Container Start Time: [%s]\n\t"\ + "Test Execution Time : [%s]" %(total_time_string, setup_time_string, testing_time_string) return summary_str