admin/splunk-security_content
1
0
Fork 0
mirror of https://github.com/splunk/security_content synced 2026-06-08 17:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Save out a json file showing the config, to include the command line arguments with credentials removed, whenever a test is actually run. This makes it trivial to reproduce the test run on another machine or again on the same machine.

Browse Source
This commit is contained in:
pyth0n1c
2021-11-23 14:30:54 -08:00
parent 914f49d6f7
commit dab746bb0c
5 changed files with 49 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py
+12 -11
View File
@@ -48,10 +48,10 @@ def copy_local_apps_to_directory(apps: dict[str,dict], target_directory)->None:
source_path = os.path.abspath(os.path.expanduser(item['local_path']))
base_name = os.path.basename(source_path)
dest_path = os.path.join(target_directory, base_name)
try:
shutil.copy(source_path, dest_path)
item['local_path'] = dest_path
print("Copied %s to apps"%(base_name))
except shutil.SameFileError as e:
# Same file, not a real error. The copy just doesn't happen
print("err:%s"%(str(e)))
@@ -153,6 +153,7 @@ def generate_escu_app(persist_security_content: bool = False) -> str:
response.raise_for_status()
with open(SPLUNK_PACKAGING_TOOLKIT_FILENAME, 'wb') as slim_file:
slim_file.write(response.content)
print("Done")
except Exception as e:
print("Error downloading the Splunk Packaging Toolkit: [%s].\n\tQuitting..." %
(str(e)), file=sys.stderr)
@@ -186,6 +187,11 @@ def generate_escu_app(persist_security_content: bool = False) -> str:
def main(args: list[str]):
try:
docker.client.from_env()
except Exception as e:
print("Error, failed to get docker client. Is Docker Running?\n\t%s"%(str(e)))
requests.packages.urllib3.disable_warnings()
start_datetime = datetime.now()
@@ -197,8 +203,8 @@ def main(args: list[str]):
elif action != "run":
print("Unsupported action: [%s]" % (action), file=sys.stderr)
sys.exit(1)
'''
parser = argparse.ArgumentParser(description="CI Detection Testing")
parser.add_argument("-b", "--branch", type=str, required=True, help="security content branch")
@@ -274,13 +280,8 @@ def main(args: list[str]):
settings['detections_list'],
settings['detections_file'])
# if len(all_test_files) == 0:
# print("No files were found to be tested. While this could be due to an error, "\
# "this could be correct if there were no changes to detections. We will "\
# "exit with success.\n\tQuitting...")
# sys.exit(0)
local_volume_absolute_path = os.path.abspath(
os.path.join(os.getcwd(), "apps"))
try:
@@ -444,7 +445,7 @@ def main(args: list[str]):
interactive_failure=settings['interactive_failure'])
print(cm.containers[0].environment)
cm.run_test()
'''
automated_detection_testing/ci/detection_testing_batch/modules/container_manager.py
+1 -1
View File
@@ -103,7 +103,7 @@ class ContainerManager:
self.baseline['TEST_FINISH_TIME'] = str(stop_time)
duration = stop_time - self.start_time
self.baseline['TEST_DURATION'] = duration - datetime.timedelta(microseconds=duration.microseconds)
self.baseline['TEST_DURATION'] = str(duration - datetime.timedelta(microseconds=duration.microseconds))
self.synchronization_object.finish(self.baseline)
automated_detection_testing/ci/detection_testing_batch/modules/github_service.py
+6 -5
View File
@@ -50,6 +50,7 @@ class GithubService:
summary_file: str = None) -> list[str]:
pruned_tests = []
csvlines = []
for detection in detections_to_prune:
if os.path.basename(detection).startswith("ssa") and exclude_ssa:
continue
@@ -61,8 +62,8 @@ class GithubService:
test_filepath_without_security_content = str(
pathlib.Path(*pathlib.Path(test_filepath).parts[1:]))
# If no types are provided, then we will get everything
if 'type' in description and (description['type'] in types_to_test or len( types_to_test) == 0):
# print(description['type'])
if 'type' in description and (description['type'] in types_to_test or len(types_to_test) == 0):
if not os.path.exists(test_filepath):
print("Detection [%s] references [%s], but it does not exist" % (
detection, test_filepath))
@@ -124,7 +125,7 @@ class GithubService:
(detections_list, detections_file), file=sys.stderr)
sys.exit(1)
elif detections_list is not None:
tests = self.get_selected_test_files(detections_list, folders, types)
tests = self.get_selected_test_files(detections_list, types)
elif detections_file is not None:
try:
with open(detections_file,'r') as f:
@@ -156,7 +157,7 @@ class GithubService:
"Anomaly", "Hunting", "TTP"],
previously_successful_tests: list[str] = []) -> list[str]:
return self.prune_detections(detection_file_list, types_to_test, previously_successful_tests)
return self.prune_detections(detection_file_list, types_to_test, previously_successful_tests)
def get_all_tests_and_detections(self,
folders: list[str] = [
@@ -170,7 +171,7 @@ class GithubService:
os.path.join("security_content/detections", folder), "*.yml"))
# Prune this down to only the subset of detections we can test
return self.prune_detections(detections, types_to_test, previously_successful_tests)
return self.prune_detections(detections, types_to_test, previously_successful_tests)
def get_all_files_in_folder(self, foldername: str, extension: str) -> list[str]:
filenames = glob.glob(os.path.join(foldername, extension))
automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py
+20 -1
View File
@@ -1,4 +1,6 @@
import argparse
import copy
import datetime
import json
from typing import OrderedDict, Union
import modules.validate_args as validate_args
@@ -86,6 +88,16 @@ def update_config_with_cli_arguments(args_dict:dict)->tuple[str, dict]:
print("Failure while processing updated settings from command line.\n\tQuitting...", file=sys.stderr)
sys.exit(1)
now = datetime.datetime.now()
configname = now.strftime('%Y-%m-%dT%H:%M:%S.%f%z') + '-test-run.json'
with open(configname,'w') as test_config:
settings_with_creds_stripped = copy.deepcopy(settings)
#strip out credentials
settings_with_creds_stripped['splunkbase_password'] = None
settings_with_creds_stripped['splunkbase_username'] = None
settings_with_creds_stripped['container_password'] = None
validate_args.validate_and_write(settings_with_creds_stripped, test_config)
return ("run", settings)
@@ -141,6 +153,13 @@ def parse(args)->tuple[str,dict]:
"if downloading packages from Splunkbase. While this can "
"be stored in the config file, it is strongly recommended "
"to enter it at runtime.")
run_parser.add_argument('-b', '--branch', required=False, type=str,
help="The branch to run the tests on.")
run_parser.add_argument('-m', '--mode', required=False, type=str,
help="The mode all, changes, or selected for the testing.")
run_parser.add_argument('-pass', '--splunkbase_password', required=False, type=str,
help="Password for login to splunkbase. This is required if "
"downloading packages from Splunkbase. While this can be "
@@ -158,7 +177,7 @@ def parse(args)->tuple[str,dict]:
"file is set to true, it will override the default False for this. True "\
"will override the default value in the config file.")
run_parser.add_argument("-m", "--mock", required=False,
run_parser.add_argument("-mock", "--mock", required=False,
action="store_true",
help="Split into multiple configs, don't actually run the tests. If the config "\
"file is set to true, it will override the default False for this. True "\
automated_detection_testing/ci/detection_testing_batch/modules/splunk_container.py
+10 -10
View File
@@ -59,7 +59,7 @@ class SplunkContainer:
self.management_port = management_port_tuple[1]
self.container = self.make_container()
self.thread = threading.Thread(target=self.run_container)
self.thread = threading.Thread(target=self.run_container, )
self.container_start_time = 0
self.test_start_time = 0
@@ -221,7 +221,7 @@ class SplunkContainer:
if self.container_start_time == -1:
total_time_string = "NOT STARTED"
else:
total_time_rounded = datetime.timedelta(
total_time_rounded = datetime.timedelta( seconds =
round(current_time - self.container_start_time))
total_time_string = str(total_time_rounded)
@@ -229,7 +229,7 @@ class SplunkContainer:
if self.test_start_time == -1 or self.container_start_time == -1:
setup_time_string = "NOT SET UP"
else:
setup_secounds_rounded = datetime.timedelta(
setup_secounds_rounded = datetime.timedelta(seconds =
round(self.test_start_time - self.container_start_time))
setup_time_string = str(setup_secounds_rounded)
@@ -237,7 +237,7 @@ class SplunkContainer:
if self.test_start_time == -1 or self.num_tests_completed == 0:
testing_time_string = "NO TESTS COMPLETED"
else:
testing_seconds_rounded = datetime.timedelta(
testing_seconds_rounded = datetime.timedelta( seconds =
round(current_time - self.test_start_time))
# Get the approximate time per test. This is a clunky way to get rid of decimal
@@ -246,13 +246,13 @@ class SplunkContainer:
timedelta_per_test_rounded = timedelta_per_test - \
datetime.timedelta(
microseconds=timedelta_per_test.microseconds)
testing_time_string = "%s per test (%d tests)"%(timedelta_per_test_rounded, self.num_tests_completed)
testing_time_string = "%s per test (%d tests)"%(timedelta_per_test_rounded, str(testing_seconds_rounded))
summary_str = "[%s] Summary\n\t"\
"Total Time :"\
"Container Start Time:"\
"Test Execution Time :" %(total_time_string, setup_time_string, testing_time_string)
summary_str = "Summary\n\t"\
"Total Time : [%s]\n\t"\
"Container Start Time: [%s]\n\t"\
"Test Execution Time : [%s]" %(total_time_string, setup_time_string, testing_time_string)
return summary_str