diff --git a/bin/docker_detection_tester/modules/splunk_sdk.py b/bin/docker_detection_tester/modules/splunk_sdk.py index cdca4f0995..75cb6b329e 100644 --- a/bin/docker_detection_tester/modules/splunk_sdk.py +++ b/bin/docker_detection_tester/modules/splunk_sdk.py @@ -7,10 +7,10 @@ import requests import time import timeit import datetime -from typing import Union +from typing import Union, Tuple DEFAULT_EVENT_HOST = "ATTACK_DATA_HOST" -DEFAULT_DATA_INDEX = "main" +DEFAULT_DATA_INDEX = set(["main"]) FAILURE_SLEEP_INTERVAL_SECONDS = 60 def enable_delete_for_admin(splunk_host:str, splunk_port:int, splunk_password:str)->bool: @@ -125,85 +125,37 @@ def wait_for_indexing_to_complete(splunk_host, splunk_port, splunk_password, sou time.sleep(check_interval_seconds) -def test_baseline_search(splunk_host, splunk_port, splunk_password, search, pass_condition, baseline_name, baseline_file, earliest_time, latest_time)->dict: - try: - service = client.connect( - host=splunk_host, - port=splunk_port, - username='admin', - password=splunk_password - ) - except Exception as e: - raise(Exception("Unable to connect to Splunk instance: " + str(e))) - - - - # search and replace \\ with \\\ - # search = search.replace('\\','\\\\') - - if search.startswith('|'): - updated_search = search - else: - updated_search = 'search ' + search - - kwargs = {"exec_mode": "blocking", - "dispatch.earliest_time": earliest_time, - "dispatch.latest_time": latest_time} - - splunk_search = updated_search + ' ' + pass_condition - - try: - job = service.jobs.create(splunk_search, **kwargs) - except Exception as e: - raise(Exception("Unable to execute baseline: " + str(e))) - - - test_results = dict() - test_results['diskUsage'] = job['diskUsage'] - test_results['runDuration'] = job['runDuration'] - test_results['baseline_name'] = baseline_name - test_results['baseline_file'] = baseline_file - test_results['scanCount'] = job['scanCount'] - - if int(job['resultCount']) != 1: - print("Test failed for baseline: " + baseline_name) - test_results['error'] = True - return test_results - else: - print("Test successful for baseline: " + baseline_name) - test_results['error'] = False - return test_results - def test_detection_search(splunk_host:str, splunk_port:int, splunk_password:str, search:str, pass_condition:str, - detection_name:str, detection_file:str, earliest_time:str, latest_time:str, attempts_remaining:int=4, - failure_sleep_interval_seconds:int=FAILURE_SLEEP_INTERVAL_SECONDS)->dict: + detection_name:str, earliest_time:str, latest_time:str, attempts_remaining:int=4, + failure_sleep_interval_seconds:int=FAILURE_SLEEP_INTERVAL_SECONDS, FORCE_ALL_TIME=True)->Tuple[bool, dict]: #Since this is an attempt, decrement the number of remaining attempts attempts_remaining -= 1 + #remove leading and trailing whitespace from the detection. + #If we don't do this with leading whitespace, this can cause + #an issue with the logic below - mainly prepending "|" in front + # of searches that look like " | tstats " + if search != search.strip(): + print(f"The detection contained in {detection_name} contains leading or trailing whitespace. Please update this search to remove that whitespace.") + search = search.strip() + if search.startswith('|'): updated_search = search else: updated_search = 'search ' + search - kwargs = {"exec_mode": "blocking", - "dispatch.earliest_time": "-1d", - "dispatch.latest_time": "now"} - splunk_search = updated_search + ' ' + pass_condition - test_results = dict() - - #These will always be present. By default, we will say that the - #test has failed AND there was an error (until they are set otherwise) - test_results['search_string'] = splunk_search - test_results['detection_name'] = detection_name - test_results['detection_file'] = detection_file - - test_results['success'] = False - test_results['error'] = True + #Set the mode and timeframe, if required + kwargs = {"exec_mode": "blocking"} + if not FORCE_ALL_TIME: + kwargs.update({"earliest_time": earliest_time, + "latest_time": latest_time}) + + #Append the pass condition to the search + splunk_search = f"{updated_search} {pass_condition}" - try: service = client.connect( host=splunk_host, @@ -215,17 +167,7 @@ def test_detection_search(splunk_host:str, splunk_port:int, splunk_password:str, except Exception as e: error_message = "Unable to connect to Splunk instance: %s"%(str(e)) print(error_message,file=sys.stderr) - test_results['error'] = True - test_results['detection_error'] = error_message - return test_results - - - # search and replace \\ with \\\ - # search = search.replace('\\','\\\\') - - - - #print("SEARCH: %s"%(splunk_search)) + return True, {"error":error_message} try: @@ -233,47 +175,16 @@ def test_detection_search(splunk_host:str, splunk_port:int, splunk_password:str, results_stream = job.results(output_mode='json') except Exception as e: - error_message = "Unable to execute detection: %s"%(str(e)) print(error_message,file=sys.stderr) - test_results['error'] = True - test_results['detection_error'] = error_message - return test_results + return True, {"error":error_message} - test_results['diskUsage'] = job['diskUsage'] - test_results['runDuration'] = job['runDuration'] - test_results['scanCount'] = job['scanCount'] + #Return all the content returned by the search + return False, job.content + + +def delete_attack_data(splunk_host:str, splunk_password:str, splunk_port:int, indices:set[str]=[DEFAULT_DATA_INDEX], host:str=DEFAULT_EVENT_HOST)->bool: - #If we get this far, then there was not an error - #The search may have FAILED, but there was no error in the search - test_results['error'] = False - - - #Should this be 1 for a pass, or should it be greater than 0? - if int(job['resultCount']) != 1: - #print("Test failed for detection: " + detection_name) - if attempts_remaining > 0: - print(f"Execution of test failed for [{detection_name}]. Sleeping for [{failure_sleep_interval_seconds} seconds] and trying up to {attempts_remaining} more times...") - time.sleep(failure_sleep_interval_seconds) - return test_detection_search(splunk_host, splunk_port, splunk_password, search, pass_condition, detection_name, detection_file, - earliest_time, latest_time, attempts_remaining=attempts_remaining, - failure_sleep_interval_seconds=failure_sleep_interval_seconds) - else: - test_results['success'] = False - return test_results - else: - #print("Test successful for detection: " + detection_name) - test_results['success'] = True - return test_results - - -def delete_attack_data(splunk_host:str, splunk_password:str, splunk_port:int, wait_on_delete:Union[dict,None], search_string:str, detection_filename:str, indices:list[str]=[DEFAULT_DATA_INDEX], host:str=DEFAULT_EVENT_HOST)->bool: - - if wait_on_delete: - print(wait_on_delete['message']) - print("FILENAME : [%s]"%(detection_filename)) - print("SEARCH :\n%s"%(search_string)) - _ = input("****************Press ENTER to Complete Test and DELETE data****************\n\n\n") try: service = client.connect( diff --git a/bin/docker_detection_tester/modules/testing_service.py b/bin/docker_detection_tester/modules/testing_service.py index 463cf3f545..7812d47764 100644 --- a/bin/docker_detection_tester/modules/testing_service.py +++ b/bin/docker_detection_tester/modules/testing_service.py @@ -1,5 +1,6 @@ import re +import shutil #import ansible_runner import yaml @@ -36,29 +37,8 @@ def test_detection_wrapper(container_name:str, splunk_ip:str, splunk_password:st - result_test, indices_to_delete = test_detection(splunk_ip, splunk_port, splunk_password, test_file, attack_data_root_folder) + result_test, indices_to_delete = test_detection(splunk_ip, splunk_port, splunk_password, test_file, attack_data_root_folder, wait_on_failure, wait_on_completion) - - if result_test is None: - #We failed so early in the process that we could not produce any meaningful result - raise(Exception("Test execution Error")) - - #enter = input("Run some tests from [%s] on [%s] - we don't delete until you hit enter :)"%(container_name, test_file)) - # delete test data - search_string = result_test['detection_result']['search_string'] - - - #search failed if there was an error or the detection failed to produce the expected result - #print("Elapsed search time: %s"%(elapsed_search_time_string)) - if (wait_on_failure or wait_on_completion) and (result_test['detection_result']['error'] or not result_test['detection_result']['success']): - wait_on_delete = {'message':"\n\n\n****SEARCH FAILURE : Allowing time to debug search/data****"} - elif wait_on_completion: - wait_on_delete = {'message':"\n\n\n****SEARCH SUCCESS : Allowing time to examine search/data****"} - else: - wait_on_delete = None - - - splunk_sdk.delete_attack_data(splunk_ip, splunk_password, splunk_port, wait_on_delete, search_string, test_file, indices = indices_to_delete) return result_test @@ -79,11 +59,11 @@ 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[dict], attack_data_folder:str)->list[dict]: +def execute_tests(splunk_ip:str, splunk_port:int, splunk_password:str, tests:list[dict], attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool)->list[dict]: results = [] for test in tests: try: - results.append(execute_test(splunk_ip, splunk_port, splunk_password, test, attack_data_folder)) + results.append(execute_test(splunk_ip, splunk_port, splunk_password, test, attack_data_folder, wait_on_failure, wait_on_completion)) except Exception as e: raise(Exception(f"Unknown error executing test: {str(e)}")) return results @@ -104,37 +84,59 @@ def execute_baseline(splunk_ip:str, splunk_port:int, splunk_password:str, baseli baseline['earliest_time'], baseline['latest_time']) return result -def execute_test(splunk_ip:str, splunk_port:int, splunk_password:str, test:dict, attack_data_folder:str)->dict: +def execute_test(splunk_ip:str, splunk_port:int, splunk_password:str, test:dict, attack_data_folder:str, wait_on_failure:bool, wait_on_completion:bool)->dict: print(f"\tExecuting test {test['name']}") - result_test = dict() + test_result = { + "name": test['name'], + "file": test['file'], + "status": False, + "logic": False, + "noise": False, + } + #replay all of the attack data test_indices = replay_attack_data_files(splunk_ip, splunk_port, splunk_password, test['attack_data'], attack_data_folder) #Run the baseline(s) if they exist for this test if 'baseline' in test: - result_test['baselines_result'] = execute_baselines(splunk_ip, splunk_port, splunk_password, test['baselines']) + test_result['baselines_result'] = execute_baselines(splunk_ip, splunk_port, splunk_password, test['baselines']) + detection = load_file(os.path.join(os.path.dirname(__file__), '../security_content/detections', test['file'])) - - detection_file_name = test['file'] - detection = load_file(os.path.join(os.path.dirname(__file__), '../security_content/detections', detection_file_name)) + error, job_result = splunk_sdk.test_detection_search(splunk_ip, splunk_port, splunk_password, detection['search'], test['pass_condition'], detection['name'], test['file'], test['earliest_time'], test['latest_time']) + if error: + test_result['message'] = job_result['message'] + return test_result + else: + #Mark whether or not the test passed + if job_result['eventCount'] == 1: + test_result["status"] = True - detection_result = splunk_sdk.test_detection_search(splunk_ip, splunk_port, splunk_password, detection['search'], test['pass_condition'], detection['name'], test['file'], test['earliest_time'], test['latest_time']) - if detection_result['error']: - print("There was an error running the search: %s"%(detection_result['search_string'])) + JOB_FIELDS = ["runDuration", "scanCount", "eventCount", "resultCount", "performance", "search"] + #Populate with all the fields we want to collect + for job_field in JOB_FIELDS: + test_result[job_field] = job_result.get(job_field, None) + if wait_on_completion or (wait_on_failure and (test_result['status'] == False)): + # The user wants to debug the test + message_template = "\n\n\n****SEARCH {status} : Allowing time to debug search/data****" + if test_result['status'] == False: + # The test failed + message_template.format(status="FAILURE") + + else: + #The test passed + message_template.format(status="SUCCESS") + + _ = input(message_template) - - detection_result['detection_name'] = test['name'] - detection_result['detection_file'] = test['file'] - result_test['detection_result'] = detection_result - - - return result_test + splunk_sdk.delete_attack_data(splunk_ip, splunk_password, splunk_port, indices = test_indices) + + return test_result def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str, attack_data_file:dict, attack_data_folder:str)->str: """Function to replay a single attack data file. Any exceptions generated during executing @@ -186,7 +188,7 @@ def replay_attack_data_file(splunk_ip:str, splunk_port:int, splunk_password:str, raise Exception("There was an error waiting for indexing to complete.") #Return the name of the index that we uploaded to - return target_index + return upload_index @@ -212,7 +214,7 @@ def replay_attack_data_files(splunk_ip:str, splunk_port:int, splunk_password:str raise(Exception(f"Error replaying attack data file {attack_data_file['file_name']}: {str(e)}")) return test_indices -def test_detection(splunk_ip:str, splunk_port:int, splunk_password:str, test_file:str, attack_data_root_folder)->list[dict]: +def test_detection(splunk_ip:str, splunk_port:int, splunk_password:str, test_file:str, attack_data_root_folder, wait_on_failure:bool, wait_on_completion:bool)->list[dict]: #Raises exception if it doesn't find the file test_file_obj = load_file(os.path.join("security_content/", test_file)) @@ -220,8 +222,11 @@ def test_detection(splunk_ip:str, splunk_port:int, splunk_password:str, test_fil abs_folder_path = mkdtemp(prefix="DATA_", dir=attack_data_root_folder) - - results = execute_tests(splunk_ip, splunk_port, splunk_password, test_file_obj['tests'], abs_folder_path) + results = execute_tests(splunk_ip, splunk_port, splunk_password, test_file_obj['tests'], abs_folder_path, wait_on_failure, wait_on_completion) + #Delete the folder and all of the data inside of it + shutil.rmtree(abs_folder_path) + + return results