From 46da0af83b22d52ff236b3eeef5657816d1c890b Mon Sep 17 00:00:00 2001 From: pyth0n1c <87383215+pyth0n1c@users.noreply.github.com> Date: Mon, 19 Sep 2022 14:47:12 -0700 Subject: [PATCH] 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. --- .../detection_testing_execution.py | 1 + .../modules/container_manager.py | 11 +- .../modules/splunk_container.py | 42 +++- .../modules/splunk_sdk.py | 2 +- .../modules/testing_service.py | 207 ++++++++++++++---- 5 files changed, 213 insertions(+), 50 deletions(-) diff --git a/bin/docker_detection_tester/detection_testing_execution.py b/bin/docker_detection_tester/detection_testing_execution.py index 786fc18c94..3b21e4d6c3 100644 --- a/bin/docker_detection_tester/detection_testing_execution.py +++ b/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'], diff --git a/bin/docker_detection_tester/modules/container_manager.py b/bin/docker_detection_tester/modules/container_manager.py index 6357edc6bd..4104d0842b 100644 --- a/bin/docker_detection_tester/modules/container_manager.py +++ b/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, diff --git a/bin/docker_detection_tester/modules/splunk_container.py b/bin/docker_detection_tester/modules/splunk_container.py index e773a97fd9..42bc420f41 100644 --- a/bin/docker_detection_tester/modules/splunk_container.py +++ b/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!") diff --git a/bin/docker_detection_tester/modules/splunk_sdk.py b/bin/docker_detection_tester/modules/splunk_sdk.py index a4875cc0ed..ddf12e0bde 100644 --- a/bin/docker_detection_tester/modules/splunk_sdk.py +++ b/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) diff --git a/bin/docker_detection_tester/modules/testing_service.py b/bin/docker_detection_tester/modules/testing_service.py index cccf72843a..7b6719f73b 100644 --- a/bin/docker_detection_tester/modules/testing_service.py +++ b/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)