Use hec, with hec setup and

token grabbing and indexer
acknowledgement, to replay
data.  Test seaches, to include
baselines, using exponential
backoff algorithm. This can result
in a huge testing speedup.

bin/docker_detection_tester/detection_testing_execution.py

@@ -527,6 +527,7 @@ def main(args: list[str]):
                                                 files_to_copy_to_container=files_to_copy_to_container,
                                                 web_port_start=8000,
                                                 management_port_start=8089,
+                                                hec_port_start=8088,
                                                 mounts=mounts,
                                                 show_container_password=settings['show_splunk_app_password'],
                                                 container_password=settings['splunk_app_password'],

bin/docker_detection_tester/modules/container_manager.py

@@ -17,6 +17,7 @@ from typing import Union
 from modules.test_objects import ResultsManager, Detection
 
 WEB_PORT_STRING = "8000/tcp"
+HEC_PORT_STRING = "8088/tcp"
 MANAGEMENT_PORT_STRING = "8089/tcp"
 
 
@@ -34,6 +35,7 @@ class ContainerManager:
         files_to_copy_to_container: OrderedDict = OrderedDict(),
         web_port_start: int = 8000,
         management_port_start: int = 8089,
+        hec_port_start: int = 8088,
         mounts: list[dict[str, str]] = [],
         show_container_password:bool=True,
         container_password: Union[str, None] = None,
@@ -77,6 +79,7 @@ class ContainerManager:
             num_containers,
             web_port_start,
             management_port_start,
+            hec_port_start,
             splunkbase_username,
             splunkbase_password,
             files_to_copy_to_container,
@@ -161,6 +164,7 @@ class ContainerManager:
         num_containers: int,
         web_port_start: int,
         management_port_start: int,
+        hec_port_start: int,
         splunkbase_username: Union[str, None] = None,
         splunkbase_password: Union[str, None] = None,
         files_to_copy_to_container: OrderedDict = OrderedDict(),
@@ -177,10 +181,8 @@ class ContainerManager:
         for index in range(num_containers):
             container_name = container_name_template % index
             web_port_tuple = (WEB_PORT_STRING, web_port_start + index)
-            management_port_tuple = (
-                MANAGEMENT_PORT_STRING,
-                management_port_start + index,
-            )
+            management_port_tuple = (MANAGEMENT_PORT_STRING, management_port_start + 2*index)
+            hec_port_tuple = (HEC_PORT_STRING, hec_port_start + 2*index)
             
             new_containers.append(
                 splunk_container.SplunkContainer(
@@ -190,6 +192,7 @@ class ContainerManager:
                     self.apps,
                     web_port_tuple,
                     management_port_tuple,
+                    hec_port_tuple,
                     self.container_password,
                     files_to_copy_to_container,
                     self.mounts,

bin/docker_detection_tester/modules/splunk_container.py

@@ -19,6 +19,7 @@ import threading
 import wrapt_timeout_decorator
 import sys
 import traceback
+import uuid
 SPLUNKBASE_URL = "https://splunkbase.splunk.com/app/%d/release/%s/download"
 SPLUNK_START_ARGS = "--accept-license"
 
@@ -33,6 +34,7 @@ class SplunkContainer:
         apps: OrderedDict,
         web_port_tuple: tuple[str, int],
         management_port_tuple: tuple[str, int],
+        hec_port_tuple: tuple[str,int],
         container_password: str,
         files_to_copy_to_container: OrderedDict = OrderedDict(),
         mounts: list[docker.types.Mount] = [],
@@ -58,9 +60,10 @@ class SplunkContainer:
         self.environment = self.make_environment(
             apps, container_password, splunkbase_username, splunkbase_password
         )
-        self.ports = self.make_ports(web_port_tuple, management_port_tuple)
+        self.ports = self.make_ports(web_port_tuple, management_port_tuple, hec_port_tuple)
         self.web_port = web_port_tuple[1]
         self.management_port = management_port_tuple[1]
+        self.hec_port = hec_port_tuple[1]
         self.container = self.make_container()
 
         self.thread = threading.Thread(target=self.run_container, )
@@ -70,6 +73,8 @@ class SplunkContainer:
         self.test_start_time = -1
         self.num_tests_completed = 0
 
+        
+
 
 
     def prepare_apps_path(
@@ -357,6 +362,37 @@ class SplunkContainer:
 
         print("Finished copying files to [%s]" % (self.container_name))
         self.wait_for_splunk_ready()
+        self.configure_hec()
+    
+    def configure_hec(self):
+        try:
+            import requests
+            auth = ('admin', self.container_password)
+            address = f"https://{self.splunk_ip}:{self.management_port}/services/data/inputs/http"
+            data = {
+                "name": "DOCKER_TEST_TESTING_HEC",
+                "index": "main",
+                "indexes": "main,_internal,_audit", #this needs to support all the indexes in test files
+                "useACK": True
+            }
+            import urllib3
+            urllib3.disable_warnings()
+            r = requests.post(address, data=data, auth=auth, verify=False)
+            if r.status_code == 201:
+                import xmltodict
+                asDict = xmltodict.parse(r.text)
+                print(asDict)
+                self.tokenString = [m['#text'] for m in asDict['feed']['entry']['content']['s:dict']['s:key'] if '@name' in m and m['@name']=='token'][0]
+                self.channel = str(uuid.uuid4())
+                print(f"WE GOT THE TOKEN STRING AND IT IS: {self.tokenString}")
+                
+            else:
+                raise(Exception(f"Error setting up hec.  Response code from {address} was [{r.status_code}]: {r.text} "))
+            
+        except Exception as e:
+            print(f"There was an issue setting up HEC.... of course.  {str(e)}")
+            _ = input("waiting here....")
+        
         
     def successfully_finish_tests(self)->None:
         try:
@@ -427,6 +463,7 @@ class SplunkContainer:
             print("Container [%s]--->[%s]" %
                   (self.container_name, str(detection_to_test.detectionFile.path)))
             try:
+                print("ONE")
                 result = testing_service.test_detection(
                     self.splunk_ip,
                     self.management_port,
@@ -434,7 +471,8 @@ class SplunkContainer:
                     detection_to_test,
                     self.synchronization_object.attack_data_root_folder,
                     wait_on_failure=self.interactive_failure,
-                    wait_on_completion = self.interactive
+                    wait_on_completion = self.interactive,
+                    container=self
                 )
                 
                 print("finished with a detection!")

bin/docker_detection_tester/modules/splunk_sdk.py

@@ -102,7 +102,7 @@ def get_number_of_indexed_events(splunk_host, splunk_port, splunk_password, inde
     
 
 
-def wait_for_indexing_to_complete(splunk_host, splunk_port, splunk_password, sourcetype:str, index:str, check_interval_seconds:int=10)->bool:
+def wait_for_indexing_to_complete(splunk_host, splunk_port, splunk_password, sourcetype:str, index:str, check_interval_seconds:int=5)->bool:
     startTime = timeit.default_timer()
     previous_count = -1
     time.sleep(check_interval_seconds)

bin/docker_detection_tester/modules/testing_service.py

@@ -1,6 +1,7 @@
 
 import re
 import shutil
+import json
 
 #import ansible_runner
 
@@ -16,7 +17,13 @@ import splunklib.client as client
 from modules.test_objects import Detection, Test, Baseline, TestResult, AttackData
 
 
+from typing import Union
+import urllib.parse
+from urllib3 import disable_warnings
+import requests
+import pathlib
 
+import os
 
 def get_service(splunk_ip:str, splunk_port:int, splunk_password:str):
 
@@ -32,19 +39,19 @@ def get_service(splunk_ip:str, splunk_port:int, splunk_password:str):
     return service
 
 
-def execute_tests(splunk_ip:str, splunk_port:int, splunk_password:str, tests:list[Test], attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool)->bool:
+def execute_tests(splunk_ip:str, splunk_port:int, splunk_password:str, tests:list[Test], attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool, container)->bool:
+    print("THREE")    
+    success = True
+    for test in tests:
+        try:
+            #Run all the tests, even if the test fails.  We still want to get the results of failed tests
+            result = execute_test(splunk_ip, splunk_port, splunk_password, test, attack_data_folder, wait_on_failure, wait_on_completion,container)
+            #And together the result of the test so that if any one test fails, it causes this function to return False                
+            success &= result
+        except Exception as e:
+            raise(Exception(f"Unknown error executing test: {str(e)}"))
+    return success
         
-        success = True
-        for test in tests:
-            try:
-                #Run all the tests, even if the test fails.  We still want to get the results of failed tests
-                result = execute_test(splunk_ip, splunk_port, splunk_password, test, attack_data_folder, wait_on_failure, wait_on_completion)
-                #And together the result of the test so that if any one test fails, it causes this function to return False                
-                success &= result
-            except Exception as e:
-                raise(Exception(f"Unknown error executing test: {str(e)}"))
-        return success
-            
 
 
 
@@ -97,22 +104,37 @@ def execute_baseline(splunk_ip:str, splunk_port:int, splunk_password:str, baseli
     
     return baseline.result.success
     
-def execute_test(splunk_ip:str, splunk_port:int, splunk_password:str, test:Test, attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool)->bool:
-    print(f"\tExecuting test {test.name}")
-    
-    #replay all of the attack data
-    test_indices = replay_attack_data_files(splunk_ip, splunk_port, splunk_password, test.attack_data, attack_data_folder)
 
+def execute_test(splunk_ip:str, splunk_port:int, splunk_password:str, test:Test, attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool,container)->bool:
     
-    
-    #Run the baseline(s) if they exist for this test
-    if execute_baselines(splunk_ip, splunk_port, splunk_password, test.baselines) is not True:
-        #One of the baselines failed. No sense in running the real test
-        test.result = TestResult(generated_exception={'message':"Baseline(s) failed"})
+    print(f"\tExecuting test {test.name}")
+    #replay all of the attack data
+    test_indices = replay_attack_data_files(splunk_ip, splunk_port, splunk_password, test.attack_data, attack_data_folder,container)
+
+    import timeit, time
+    start = timeit.default_timer()
+    MAX_TIME = 120
+    sleep_base = 2
+    sleep_exp = 0
+    while True:
+        sleeptime = sleep_base**sleep_exp
+        sleep_exp += 1
+        print(f"Sleep for {sleeptime} for ingest") 
+        time.sleep(sleeptime)
+        #Run the baseline(s) if they exist for this test
+        if execute_baselines(splunk_ip, splunk_port, splunk_password, test.baselines) is not True:
+            #One of the baselines failed. No sense in running the real test
+            test.result = TestResult(generated_exception={'message':"Baseline(s) failed"})
+            
+            
+        else:
+            test.result = splunk_sdk.test_detection_search(splunk_ip, splunk_port, splunk_password, test.detectionFile.search, test.pass_condition, test.name, test.earliest_time, test.latest_time)
         
-        
-    else:
-        test.result = splunk_sdk.test_detection_search(splunk_ip, splunk_port, splunk_password, test.detectionFile.search, test.pass_condition, test.name, test.earliest_time, test.latest_time)
+        if test.result.success:
+            #We were successful, no need to run again.
+            break
+        elif timeit.default_timer() - start > MAX_TIME:
+            break
         
  
     if wait_on_completion or (wait_on_failure and (test.result.success == False)):
@@ -136,7 +158,110 @@ def execute_test(splunk_ip:str, splunk_port:int, splunk_password:str, test:Test,
     #Return whether the test passed or failed
     return test.result.success
 
-def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str, attackData:AttackData, attack_data_folder:str)->str:
+
+def hec_raw_replay(base_url:str, token:str, filePath:pathlib.Path, 
+                   source:Union[str,None]=None, sourcetype:Union[str,None]=None, 
+                   host:Union[str,None]=None, channel:Union[str,None]=None, 
+                   use_https:bool=True, port:int=8088, verify=False, 
+                   path:str="services/collector/raw", wait_for_ack:bool=True):
+    
+    if verify is False:
+        #need this, otherwise every request made with the requests module
+        #and verify=False will print an error to the command line
+        disable_warnings()
+
+
+    #build the headers
+    if token.startswith('Splunk '):
+        headers = {"Authorization": token} 
+    else:
+        headers = {"Authorization": f"Splunk {token}"} #token must begin with 'Splunk 
+    
+    if channel is not None:
+        headers['X-Splunk-Request-Channel'] = channel
+    
+    
+    #Now build the URL parameters
+    url_params_dict = {}
+    if source is not None:
+        url_params_dict['source'] = source 
+    if sourcetype is not None:
+        url_params_dict['sourcetype'] = sourcetype
+    if host is not None:
+        url_params_dict['host'] = host 
+    
+    
+    if base_url.lower().startswith('http://') and use_https is True:
+        raise(Exception(f"URL {base_url} begins with http://, but use_http is {use_https}. "\
+                         "Unless you have modified the HTTP Event Collector Configuration, it is probably enabled for https only."))
+    if base_url.lower().startswith('https://') and use_https is False:
+        raise(Exception(f"URL {base_url} begins with https://, but use_http is {use_https}. "\
+                         "Unless you have modified the HTTP Event Collector Configuration, it is probably enabled for https only."))
+    
+    if not (base_url.lower().startswith("http://") or base_url.lower().startswith('https://')):
+        if use_https:
+            prepend = "https://"
+        else:
+            prepend = "http://"
+        old_url = base_url
+        base_url = f"{prepend}{old_url}"
+        print(f"Warning, the URL you provided {old_url} does not start with http:// or https://.  We have added {prepend} to convert it into {base_url}")
+    
+
+    #Generate the full URL, including the host, the path, and the params.
+    #We can be a lot smarter about this (and pulling the port from the url, checking 
+    # for trailing /, etc, but we leave that for the future)
+    url_with_path = urllib.parse.urljoin(f"{base_url}:{port}", path)
+    with open(filePath,"rb") as datafile:
+        rawData = datafile.read()
+
+    try:
+        res = requests.post(url_with_path,params=url_params_dict, data=rawData, allow_redirects = True, headers=headers, verify=verify)
+        print(f"POST Sent with return code: {res.status_code}")
+        jsonResponse = json.loads(res.text)
+    except Exception as e:
+        raise(Exception(f"There was an exception in the post: {str(e)}"))
+    
+
+    if wait_for_ack:
+        if channel is None:
+            raise(Exception("HEC replay WAIT_FOR_ACK is enabled but CHANNEL is None. Channel must be supplied to wait on ack"))
+        
+        if "ackId" not in jsonResponse:
+            raise(Exception(f"key 'ackID' not present in response from HEC server: {jsonResponse}"))
+        ackId = jsonResponse['ackId']
+        url_with_path = urllib.parse.urljoin(f"{base_url}:{port}", "services/collector/ack")
+        import timeit, time
+        start = timeit.default_timer()
+        j = {"acks":[jsonResponse['ackId']]}
+        while True:            
+            try:
+                
+                res = requests.post(url_with_path, json=j, allow_redirects = True, headers=headers, verify=verify)
+                print(f"ACKID POST Sent with return code: {res.status_code}")
+                jsonResponse = json.loads(res.text)
+                print(f"the type of ackid is {type(ackId)}")
+                if 'acks' in jsonResponse and str(ackId) in jsonResponse['acks']:
+                    if jsonResponse['acks'][str(ackId)] is True:
+                        break
+                    else:
+                        print("Waiting for ackId")
+
+                        time.sleep(1)
+
+                else:
+                    print(url_with_path)
+                    print(j)
+                    print(headers)
+                    raise(Exception(f"Proper ackID structure not found for ackID {ackId} in {jsonResponse}"))
+            except Exception as e:
+                raise(Exception(f"There was an exception in the post: {str(e)}"))
+            
+
+
+
+
+def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str, attackData:AttackData, attack_data_folder:str,container)->str:
     """Function to replay a single attack data file. Any exceptions generated during executing
     are intentionally not caught so that they can be caught by the caller.
 
@@ -151,10 +276,10 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str,
         str: index that the attack data has been replayed into on the splunk server
     """
     #Get the index we should replay the data into
-    print("replaying single attack data file")
+    
     
     descriptor, data_file = mkstemp(prefix="ATTACK_DATA_FILE_", dir=attack_data_folder)
-    if not attackData.data.startswith("https://"):
+    if not (attackData.data.startswith("https://") or attackData.data.startswith("http://")):
         #raise(Exception(f"Attack Data File {attack_data_file['file_name']} does not start with 'https://'. "  
         #                 "In the future, we will add support for non https:// hosted files, such as local files or other files. But today this is an error."))
         
@@ -168,7 +293,6 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str,
     
     else:
         #Download the file
-        print(f"download from {attackData.data}-->{data_file}")
         #We need to overwrite the file - mkstemp will create an empty file with the 
         #given name
         utils.download_file_from_http(attackData.data, data_file, overwrite_file=True) 
@@ -176,10 +300,7 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str,
     # Update timestamps before replay
     if attackData.update_timestamp:
         data_manipulation = DataManipulation()
-        print(f"ABSOLUTE FILE PATH: {data_file}")
-        import os
         relpath = os.path.relpath(data_file)
-        print(f"RELATIVE FILE PATH: {relpath}")
         data_manipulation.manipulate_timestamp(relpath, attackData.sourcetype,attackData.source)    
 
     #Get an session from the API
@@ -188,13 +309,13 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str,
     upload_index = service.indexes[attackData.index]
         
     #Upload the data
-    print(f"the data file is: {data_file}")
-    with open(data_file, 'rb') as target:
-        upload_index.submit(target.read(), sourcetype=attackData.sourcetype, source=attackData.source, host=splunk_sdk.DEFAULT_EVENT_HOST)
+    hec_raw_replay(container.splunk_ip, container.tokenString, pathlib.Path(data_file), attackData.source, attackData.sourcetype, splunk_sdk.DEFAULT_EVENT_HOST, channel=container.channel)
+    
 
     #Wait for the indexing to finish
-    if not splunk_sdk.wait_for_indexing_to_complete(splunk_ip, splunk_port, splunk_password, attackData.sourcetype, upload_index):
-        raise Exception("There was an error waiting for indexing to complete.")
+    print("skip waiting for ingest since we have checked the ackid")
+    #if not splunk_sdk.wait_for_indexing_to_complete(splunk_ip, splunk_port, splunk_password, attackData.sourcetype, upload_index):
+    #    raise Exception("There was an error waiting for indexing to complete.")
     
     print('done waiting')
     #Return the name of the index that we uploaded to
@@ -205,7 +326,7 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str,
 
     
 
-def replay_attack_data_files(splunk_ip:str, splunk_port:int, splunk_password:str, attackDataObjects:list[AttackData], attack_data_folder:str)->set[str]:
+def replay_attack_data_files(splunk_ip:str, splunk_port:int, splunk_password:str, attackDataObjects:list[AttackData], attack_data_folder:str,container)->set[str]:
     """Replay all attack data files into a splunk server as part of testing a detection. Note that this does not catch
     any exceptions, they should be handled by the caller
 
@@ -216,20 +337,20 @@ def replay_attack_data_files(splunk_ip:str, splunk_port:int, splunk_password:str
         attack_data_files (list[dict]): A list of dicts containing information about the attack data file
         attack_data_folder (str): The folder for downloaded or copied attack data to reside
     """
-    print('replaying all attack data')
     test_indices = set()
     for attack_data_file in attackDataObjects:
         try:
-            test_indices.add(replay_attack_data_file(splunk_ip, splunk_port, splunk_password, attack_data_file, attack_data_folder))
+            test_indices.add(replay_attack_data_file(splunk_ip, splunk_port, splunk_password, attack_data_file, attack_data_folder,container))
         except Exception as e:
             raise(Exception(f"Error replaying attack data file {attack_data_file.data}: {str(e)}"))
     return test_indices
 
-def test_detection(splunk_ip:str, splunk_port:int, splunk_password:str, detection:Detection, attack_data_root_folder, wait_on_failure:bool, wait_on_completion:bool)->bool:
-    
+
+def test_detection(splunk_ip:str, splunk_port:int, splunk_password:str, detection:Detection, attack_data_root_folder, wait_on_failure:bool, wait_on_completion:bool,container)->bool:
+    print("TWO")
 
     abs_folder_path = mkdtemp(prefix="DATA_", dir=attack_data_root_folder)
-    success = execute_tests(splunk_ip, splunk_port, splunk_password, detection.testFile.tests, abs_folder_path, wait_on_failure, wait_on_completion)
+    success = execute_tests(splunk_ip, splunk_port, splunk_password, detection.testFile.tests, abs_folder_path, wait_on_failure, wait_on_completion, container)
     detection.get_detection_result()
     #Delete the folder and all of the data inside of it
     #shutil.rmtree(abs_folder_path)