diff --git a/bin/docker_detection_tester/detection_testing_execution.py b/bin/docker_detection_tester/detection_testing_execution.py index 4d9f1898f2..990d6a8d66 100644 --- a/bin/docker_detection_tester/detection_testing_execution.py +++ b/bin/docker_detection_tester/detection_testing_execution.py @@ -29,9 +29,10 @@ import requests.packages.urllib3 from docker.client import DockerClient from requests import get -import modules.new_arguments2 + + from modules import (container_manager, new_arguments2, - testing_service, validate_args) + testing_service, validate_args, utils) from modules.github_service import GithubService from modules.validate_args import validate, validate_and_write, ES_APP_NAME @@ -56,17 +57,6 @@ MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING = 2 -def download_file_from_http(url:str, destination_file:str, overwrite_file:bool=False)->None: - if os.path.exists(destination_file) and overwrite_file is False: - print(f"[{destination_file}] already exists...using cached version") - return - print(f"downloading to [{destination_file}]") - file_to_download = requests.get(url, stream=True) - with open(destination_file, "wb") as output: - for piece in file_to_download.iter_content(chunk_size=(1024*1024)): - output.write(piece) - - def copy_local_apps_to_directory(apps: dict[str, dict], splunkbase_username:tuple[str,None] = None, splunkbase_password:tuple[str,None] = None, mock:bool = False, target_directory:str = "apps") -> str: if mock is True: target_directory = os.path.join("prior_config", target_directory) @@ -128,7 +118,7 @@ def copy_local_apps_to_directory(apps: dict[str, dict], splunkbase_username:tupl path_after_host = url_parse_obj[2].rstrip('/') #removes / at the end, if applicable base_name = path_after_host.rpartition('/')[-1] #just get the file name dest_path = os.path.join(target_directory, base_name) #write the whole path - download_file_from_http(http_path, dest_path) + utils.download_file_from_http(http_path, dest_path) #we need to update the local path because this is used to copy it into the container later item['local_path'] = dest_path #Remove the HTTP Path, we will use the local_path instead @@ -341,7 +331,7 @@ def main(args: list[str]): start_datetime = datetime.now() - action, settings = modules.new_arguments2.parse(args) + action, settings = new_arguments2.parse(args) if action == "configure": # Done, nothing else to do print("Configuration complete!") diff --git a/bin/docker_detection_tester/modules/splunk_container.py b/bin/docker_detection_tester/modules/splunk_container.py index 1c2c3c7f28..52c07ba832 100644 --- a/bin/docker_detection_tester/modules/splunk_container.py +++ b/bin/docker_detection_tester/modules/splunk_container.py @@ -455,8 +455,18 @@ class SplunkContainer: #pdb.set_trace() # Fill in all the "Empty" fields with default values. Otherwise, we will not be able to # process the result correctly. + detection_to_test.replace("security_content/tests", "security_content/detections") + try: + test_file_obj = testing_service.load_file(os.path.join("security_content/", detection_to_test)) + if 'file' not in test_file_obj: + raise Exception(f"'file' field not found in {detection_to_test}") + except: + test_file_obj['file'] = detection_to_test.replace("tests/", "").replace(".test.yml", ".yml") + print(f"Error getting the detection file associated with the test file. We will try our best to convert it: {detection_to_test}-->{test_file_obj['file']}") + + self.synchronization_object.addError( - {"detection_file": detection_to_test, + {"detection_file": test_file_obj['file'], "detection_error": str(e)}, duration_string = datetime.timedelta(seconds=round(timeit.default_timer() - current_test_start_time)) diff --git a/bin/docker_detection_tester/modules/testing_service.py b/bin/docker_detection_tester/modules/testing_service.py index 00fd03d3e8..72b39cfc69 100644 --- a/bin/docker_detection_tester/modules/testing_service.py +++ b/bin/docker_detection_tester/modules/testing_service.py @@ -9,6 +9,7 @@ import os import time import requests from modules.DataManipulation import DataManipulation +from modules import utils from modules import splunk_sdk import timeit from typing import Union, Tuple @@ -103,12 +104,10 @@ def test_detection(splunk_ip:str, splunk_port:int, container_name:str, splunk_pa data_upload_index = splunk_sdk.DEFAULT_DATA_INDEX indices_to_delete.add(data_upload_index) - - r = requests.get(url, allow_redirects=True) + target_file = os.path.join(folder_name, attack_data['file_name']) - with open(target_file, 'wb') as target: - target.write(r.content) - #print(target_file) + utils.download_file_from_http(url, target_file) + # Update timestamps before replay diff --git a/bin/docker_detection_tester/modules/utils.py b/bin/docker_detection_tester/modules/utils.py new file mode 100644 index 0000000000..c818e44bc0 --- /dev/null +++ b/bin/docker_detection_tester/modules/utils.py @@ -0,0 +1,15 @@ +import os +import requests + + +def download_file_from_http(url:str, destination_file:str, overwrite_file:bool=False, chunk_size=1024*1024)->None: + if os.path.exists(destination_file) and overwrite_file is False: + print(f"[{destination_file}] already exists...using cached version") + return + print(f"downloading to [{destination_file}]") + file_to_download = requests.get(url, stream=True) + if file_to_download.status_code != 200: + raise Exception(f"Error downloading the file {url}: Status Code {file_to_download.status_code}") + with open(destination_file, "wb") as output: + for piece in file_to_download.iter_content(chunk_size=chunk_size): + output.write(piece) \ No newline at end of file