Fixed output of manifest for detection_failure_manifest.json. Previously, it contained mostly default settings and most notably did not container the proper apps configuration. Also updated a few documentation strings.

bin/automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py

@@ -370,8 +370,11 @@ def main(args: list[str]):
     #sys.exit(0)
     # Make a backup of this config containing the hash and stripped credentials.
     # This makes the test perfectly reproducible.
-    validate_args.validate_and_write(
-        settings, output_file=None, strip_credentials=True)
+    reproduce_test_config, _ = validate_args.validate_and_write(settings, output_file=None, strip_credentials=True)
+    if reproduce_test_config == None:
+        print("Error - there was an error writing out the file to reproduce the test.  This should not happen, as all "\
+              "settings should have been validated by this point.\n\tQuitting...",file=sys.stderr)
+        sys.exit(1)
 
     try:
         all_test_files = github_service.get_test_files(settings['mode'],
@@ -454,6 +457,7 @@ def main(args: list[str]):
                                             settings['splunkbase_apps'],
                                             settings['branch'],
                                             settings['commit_hash'],
+                                            reproduce_test_config,
                                             files_to_copy_to_container=files_to_copy_to_container,
                                             web_port_start=8000,
                                             management_port_start=8089,
@@ -468,7 +472,7 @@ def main(args: list[str]):
 
     result = cm.run_test()
     
-    github_service.update_and_commit_passed_tests(cm.synchronization_object.successes)
+    #github_service.update_and_commit_passed_tests(cm.synchronization_object.successes)
     
 
     #Return code indicates whether testing succeeded and all tests were run.

bin/automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py

@@ -29,6 +29,7 @@ class ContainerManager:
         splunkbase_apps:OrderedDict,
         branch:str,
         commit_hash:str,
+        summarization_reproduce_failure_config:dict,
         files_to_copy_to_container: OrderedDict = OrderedDict(),
         web_port_start: int = 8000,
         management_port_start: int = 8089,
@@ -43,7 +44,7 @@ class ContainerManager:
 
     ):
         self.synchronization_object = test_driver.TestDriver(
-            test_list, num_containers)
+            test_list, num_containers, summarization_reproduce_failure_config)
 
         self.mounts = self.create_mounts(mounts)
         self.local_apps = local_apps
@@ -55,7 +56,7 @@ class ContainerManager:
             self.container_password = container_password
         
         print("\n\n***********************")
-        print("Log into your Splunk Container(s) after they boot at at http://127.0.0.1:[%d-%d]"%(web_port_start, web_port_start + num_containers - 1))
+        print("Log into your [%d] Splunk Container(s) after they boot at http://127.0.0.1:[%d-%d]"%(num_containers, web_port_start, web_port_start + num_containers - 1))
         print("\tSplunk App Username: [%s]"%("admin"))
         print("\tSplunk App Password: ", end='')
         if show_container_password:

bin/automated_detection_testing/ci/detection_testing_batch/modules/test_driver.py

@@ -12,9 +12,9 @@ import time
 import timeit
 from typing import Union
 import sys
-
+import copy
 class TestDriver:
-    def __init__(self, tests:list[str], num_containers:int):
+    def __init__(self, tests:list[str], num_containers:int, summarization_reproduce_failure_config:dict):
         #Create the queue and enque all of the tests
         self.testing_queue = queue.Queue()
         for test in tests:
@@ -35,8 +35,12 @@ class TestDriver:
         #Just make a random folder to store attack data that we donwload
         self.attack_data_root_folder = tempfile.mkdtemp(prefix="attack_data_", dir=os.getcwd())
         print("Attack data for this run will be stored at: [%s]"%(self.attack_data_root_folder))
+        
+        #Not used right now, but we will keep it around for a bit in case we want to use it again
         self.start_barrier = threading.Barrier(num_containers)
 
+        #The config that will be used for writing out the error config reproduction fiel
+        self.summarization_reproduce_failure_config = copy.deepcopy(summarization_reproduce_failure_config)
 
     def checkContainerFailure(self)->bool:
         
@@ -184,11 +188,14 @@ class TestDriver:
         res |= self.outputResultsFile(fields, os.path.join(results_directory, "combined"), combined_data, baseline)
         
         try:
-            success, test_count,pass_count,fail_count,error_count = summarize_json.outputResultsJSON("summary.json", combined_data, baseline, output_folder=results_directory)
+            success, test_count,pass_count,fail_count,error_count = \
+                summarize_json.outputResultsJSON("summary.json", combined_data, 
+                                                 baseline, output_folder=results_directory, 
+                                                 summarization_reproduce_failure_config=self.summarization_reproduce_failure_config)
             summarize_json.print_summary(test_count, pass_count, fail_count, error_count)
             res |= success
         except Exception as e:
-            print("Failure writing the summary file: [%s]",file=sys.stderr)
+            print("Failure writing the summary file: [%s]"%str(e),file=sys.stderr)
             res = False
 
         return res

bin/automated_detection_testing/ci/detection_testing_batch/summarize_json.py

@@ -6,11 +6,11 @@ import json
 from modules import validate_args
 import os.path
 from operator import itemgetter
-
+import copy
 
 def outputResultsJSON(output_filename:str, data:list[dict], baseline:OrderedDict, 
                       failure_manifest_filename = "detection_failure_manifest.json", 
-                      output_folder:str="")->tuple[bool,int,int,int,int]:
+                      output_folder:str="", summarization_reproduce_failure_config:dict={})->tuple[bool,int,int,int,int]:
     success = True
     
     try:
@@ -61,9 +61,11 @@ def outputResultsJSON(output_filename:str, data:list[dict], baseline:OrderedDict
         fail_list = [os.path.join("security_content/detections",x['detection_file'] ) for x in data_sorted if x['success'] == False]
 
         if len(fail_list) > 0:
-            failures_test_override = {"detections_list": fail_list, "interactive_failure":True, 
+                                    
+            failures_test_override = copy.deepcopy(summarization_reproduce_failure_config)
+            failures_test_override.update({"detections_list": fail_list, "no_interactive_failure":False, 
                                     "num_containers":1, "branch": baseline["branch"], "commit_hash":baseline["commit_hash"], 
-                                    "mode":"selected", "show_splunk_app_password": True}
+                                    "mode":"selected", "show_splunk_app_password": True})
             with open(os.path.join(output_folder,failure_manifest_filename),"w") as failures:
                 validate_args.validate_and_write(failures_test_override, failures)
     except Exception as e: