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

Branch was auto-updated.

Browse Source
This commit is contained in:
pyth0n1c
2022-03-23 15:21:40 -07:00
committed by GitHub
parent 054496f5b4 3b5284ff99
commit af7435aed0
7 changed files with 138 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/docker_detection_tester/detection_testing_execution.py
+1 -2
View File
@@ -435,8 +435,7 @@ def main(args: list[str]):
all_test_files = github_service.get_test_files(settings['mode'],
settings['folders'],
settings['types'],
settings['detections_list'],
settings['detections_file'])
settings['detections_list'])
#We randomly shuffle this because there are likely patterns in searches. For example,
#cloud/endpoint/network likely have different impacts on the system. By shuffling,
bin/docker_detection_tester/modules/github_service.py
+132 -79
View File
@@ -21,10 +21,15 @@ LOGGER = logging.getLogger(__name__)
SECURITY_CONTENT_URL = "https://github.com/splunk/security_content"
DETECTION_ROOT_PATH = "security_content/detections"
TEST_ROOT_PATH = "security_content/tests"
DETECTION_FILE_EXTENSION = ".yml"
TEST_FILE_EXTENSION = ".test.yml"
SSA_PREFIX = "ssa___"
class GithubService:
def __init__(self, security_content_branch: str, commit_hash: Union[str,None], PR_number: int = None, persist_security_content: bool = False):
def __init__(self, security_content_branch: str, commit_hash: Union[str,None], PR_number: Union[int,None] = None, persist_security_content: bool = False):
self.security_content_branch = security_content_branch
if persist_security_content:
@@ -81,6 +86,8 @@ class GithubService:
self.commit_hash = commit_hash
def update_and_commit_passed_tests(self, results:list[dict])->bool:
@@ -140,16 +147,17 @@ class GithubService:
def prune_detections(self,
detections_to_prune: list[str],
detection_files: list[str],
types_to_test: list[str],
previously_successful_tests: list[str],
exclude_ssa: bool = True,
summary_file: str = None) -> list[str]:
exclude_ssa: bool = True) -> list[str]:
pruned_tests = []
csvlines = []
found_error = False
for detection in detections_to_prune:
if os.path.basename(detection).startswith("ssa") and exclude_ssa:
for detection in detection_files:
if os.path.basename(detection).startswith(SSA_PREFIX) and exclude_ssa:
continue
with open(detection, "r") as d:
description = yaml.safe_load(d)
@@ -164,84 +172,112 @@ class GithubService:
if not os.path.exists(test_filepath):
print("Detection [%s] references [%s], but it does not exist" % (
detection, test_filepath))
found_error = True
#raise(Exception("Detection [%s] references [%s], but it does not exist"%(detection, test_filepath)))
elif test_filepath_without_security_content in previously_successful_tests:
print(
"Ignoring test [%s] before it has already passed previously" % (detection))
else:
# remove leading security_content/ from path
pruned_tests.append(
test_filepath_without_security_content)
if summary_file is not None:
try:
mitre_id = str(
description['tags']['mitre_attack_id'])
except:
mitre_id = 'NONE'
try:
csvlines.append({'name': description['name'], 'filename': detection, 'description': description['description'],
'search': description['search'], 'mitre_attack_id': mitre_id, 'security_domain': description['tags']['security_domain'],
'Relevant': '', 'Comments': '', "Runnable on SSA?": str(os.path.basename(detection).startswith("ssa"))})
except Exception as e:
print("Error outputting summary for [%s]: [%s]" % (
detection, str(e)))
pruned_tests.append(test_filepath_without_security_content)
else:
# Don't do anything with these files
pass
if not self.ensure_paired_detection_and_test_files([], [os.path.join("security_content", p) for p in pruned_tests], exclude_ssa):
raise(Exception("Missing one or more test/detection files. Please see the output above."))
return pruned_tests
if summary_file is not None:
print("writing")
with open(summary_file, 'w') as csvfile:
fieldnames = ['name', 'filename', 'description', 'search', 'mitre_attack_id',
'security_domain', 'Runnable on SSA?', 'Relevant', 'Comments']
writer = csv.DictWriter(
csvfile, fieldnames=fieldnames, quoting=csv.QUOTE_ALL)
writer.writeheader()
for r in csvlines:
writer.writerow(r)
def ensure_paired_detection_and_test_files(self, detection_files: list[str], test_files: list[str], exclude_ssa: bool = True)->bool:
'''
The security_content repo contains two folders: detections and test.
For EVERY detection in the detections folder, there must be a test.
for EVERY test in the tests folder, there MUST be a detection.
If this requirement is not met, then throw an error
'''
MISSING_TEMPLATE = "Missing {type} file:"\
"\n\tEXISTS - {exists}"\
"\n\tMISSING - {missing}"\
no_missing_files = True
#Check that all detection files have a test file
for detection_file in detection_files:
test_file = self.convert_detection_filename_into_test_filename(detection_file)
if not os.path.exists(test_file):
if os.path.basename(detection_file).startswith(SSA_PREFIX) and exclude_ssa is True:
print(MISSING_TEMPLATE.format(type="test", exists=detection_file, missing=test_file))
print("\tSince exclude_ssa is TRUE, this is not an error, just a warning")
else:
print(MISSING_TEMPLATE.format(type="test", exists=detection_file, missing=test_file))
no_missing_files = False
#Check that all test files have a detection file
for test_file in test_files:
detection_file = self.convert_test_filename_into_detection_filename(test_file)
if not os.path.exists(detection_file):
if os.path.basename(test_file).startswith(SSA_PREFIX) and exclude_ssa is True:
print(MISSING_TEMPLATE.format(type="detection", exists=test_file, missing=detection_file))
print("\tSince exclude_ssa is TRUE, this is not an error, just a warning")
else:
print(MISSING_TEMPLATE.format(type="detection", exists=test_file, missing=detection_file))
no_missing_files = False
return no_missing_files
def convert_detection_filename_into_test_filename(self, detection_filename:str) ->str:
head, tail = os.path.split(detection_filename)
assert head.startswith(DETECTION_ROOT_PATH), \
f"Error - Expected detection filename to start with [{DETECTION_ROOT_PATH}] but instead got {detection_filename}"
updated_head = head.replace(DETECTION_ROOT_PATH, TEST_ROOT_PATH, 1)
assert tail.endswith(DETECTION_FILE_EXTENSION),\
f"Error - Expected detection filename to end with [{DETECTION_FILE_EXTENSION}] but instead got [{detection_filename}]"
updated_tail = TEST_FILE_EXTENSION.join(tail.rsplit(DETECTION_FILE_EXTENSION))
return os.path.join(updated_head, updated_tail)
def convert_test_filename_into_detection_filename(self, test_filename:str) ->str :
head, tail = os.path.split(test_filename)
assert head.startswith(TEST_ROOT_PATH), \
f"Error - Expected test filename to start with [{TEST_ROOT_PATH}] but instead got {test_filename}"
updated_head = head.replace(TEST_ROOT_PATH, DETECTION_ROOT_PATH, 1)
assert tail.endswith(TEST_FILE_EXTENSION), \
f"Error - Expected test filename to end with [{TEST_FILE_EXTENSION}] but instead got [{test_filename}]"
updated_tail = DETECTION_FILE_EXTENSION.join(tail.rsplit(TEST_FILE_EXTENSION))
return os.path.join(updated_head, updated_tail)
if found_error == False:
return pruned_tests
else:
raise(Exception("Error(s) in processing getting detections to test: see output above for specific error information."))
def get_test_files(self, mode: str, folders: list[str], types: list[str],
detections_list: Union[list[str], None],
detections_file=Union[str, None]) -> list[str]:
detections_list: Union[list[str], None]) -> list[str]:
#Every test should have a detection associated with it. It is NOT necessarily
#true that all detections should have a test associated with them. For example,
#only certain types of detections should have a test associated with them.
self.verify_all_tests_have_detections(folders, types)
if mode == "changes":
tests = self.get_changed_test_files(folders, types)
elif mode == "selected":
if detections_list is None and detections_file is None:
if detections_list is None:
# It's actually valid to supply an EMPTY list of files and the test should pass.
# This can occur when we try to test, for example, 1 detection but start 2 containers.
# We still want this to pass testing, so we shouldn't fail there!
print(
"Trying to test a list of files, but None were provided", file=sys.stderr)
print("Trying to test a list of files, but None were provided", file=sys.stderr)
sys.exit(1)
elif detections_list is not None and detections_file is not None:
print("Both detections_list [%s] and detections_file [%s] were provided. "
"Because these confilect, we cannot test.\n\tQuitting..." %
(detections_list, detections_file), file=sys.stderr)
sys.exit(1)
elif detections_list is not None:
tests = self.get_selected_test_files(detections_list, types)
elif detections_file is not None:
try:
with open(detections_file, 'r') as f:
data = f.readlines()
# Strip all whitespace from lines and exclude lines that are just whitespace
files_to_test = [line.strip()
for line in data if len(line.strip()) > 0]
except Exception as e:
print("There was an error reading the input file [%s]: [%s].\n\t"
"Quitting..." % (detections_file, str(e)))
sys.exit(1)
tests = self.get_selected_test_files(
files_to_test, folders, types)
else:
# impossible to get here
print(
@@ -255,35 +291,53 @@ class GithubService:
"Error, unsupported mode [%s]. Mode must be one of %s", file=sys.stderr)
sys.exit(1)
return tests
def get_selected_test_files(self,
detection_file_list: list[str],
types_to_test: list[str] = [
"Anomaly", "Hunting", "TTP"],
previously_successful_tests: list[str] = []) -> list[str]:
"Anomaly", "Hunting", "TTP"]) -> list[str]:
return self.prune_detections(detection_file_list, types_to_test)
def verify_all_tests_have_detections(self, folders: list[str] = [
'endpoint', 'cloud', 'network'],
types_to_test: list[str] = [
"Anomaly", "Hunting", "TTP"],
exclude_ssa:bool=True)->bool:
all_tests = []
for folder in folders:
#Get all the tests in a folder
tests = self.get_all_files_in_folder(os.path.join(TEST_ROOT_PATH, folder), "*")
#Convert all of those tests to detection paths
for test in tests:
all_tests.append(test)
if not self.ensure_paired_detection_and_test_files([], all_tests, exclude_ssa):
raise(Exception("Missing one or more detection files. Please see the output above."))
return True
return self.prune_detections(detection_file_list, types_to_test, previously_successful_tests)
def get_all_tests_and_detections(self,
folders: list[str] = [
'endpoint', 'cloud', 'network'],
types_to_test: list[str] = [
"Anomaly", "Hunting", "TTP"],
previously_successful_tests: list[str] = []) -> list[str]:
"Anomaly", "Hunting", "TTP"]) -> list[str]:
detections = []
for folder in folders:
detections.extend(self.get_all_files_in_folder(
os.path.join("security_content/detections", folder), "*.yml"))
detections.extend(self.get_all_files_in_folder(os.path.join(DETECTION_ROOT_PATH, folder), "*"))
# 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)
def get_all_files_in_folder(self, foldername: str, extension: str) -> list[str]:
filenames = glob.glob(os.path.join(foldername, extension))
return filenames
def get_changed_test_files(self, folders=['endpoint', 'cloud', 'network'], types_to_test=["Anomaly", "Hunting", "TTP"], previously_successful_tests=[]) -> list[str]:
def get_changed_test_files(self, folders=['endpoint', 'cloud', 'network'], types_to_test=["Anomaly", "Hunting", "TTP"]) -> list[str]:
branch1 = self.security_content_branch
branch2 = 'develop'
@@ -370,7 +424,7 @@ class GithubService:
changed_detection_files.append(name)
return self.prune_detections(changed_detection_files, types_to_test, previously_successful_tests)
return self.prune_detections(changed_detection_files, types_to_test)
#detections_to_test,_,_ = self.filter_test_types(changed_detection_files)
# for f in detections_to_test:
@@ -414,4 +468,3 @@ class GithubService:
import time
time.sleep(5)
return files_to_test, files_not_to_test, error_files
bin/docker_detection_tester/modules/validate_args.py
+5 -17
View File
@@ -43,13 +43,6 @@ setup_schema = {
},
"default": None,
},
"detections_file": {
"type": ["string", "null"],
"default": None
},
"apps": {
"type": "object",
"additionalProperties": False,
@@ -259,16 +252,11 @@ def check_dependencies(settings: dict, skip_password_accessibility_check:bool=Tr
if settings['mode'] == 'selected':
# Make sure that exactly one of the following fields is populated
if settings['detections_file'] == None and settings['detections_list'] == None:
print("Error - mode was 'selected' but no detections_list or detections_file were supplied.", file=sys.stderr)
if settings['detections_list'] == None:
print("Error - mode was 'selected' but no detections_list was supplied.", file=sys.stderr)
error_free = False
elif settings['detections_file'] != None and settings['detections_list'] != None:
print("Error - mode was 'selected' but detections_list and detections_file were supplied.", file=sys.stderr)
error_free = False
if settings['mode'] != 'selected' and settings['detections_file'] != None:
print("Error - mode was not 'selected' but detections_file was supplied.", file=sys.stderr)
error_free = False
elif settings['mode'] != 'selected' and settings['detections_list'] != None:
if settings['mode'] != 'selected' and settings['detections_list'] != None:
print("Error - mode was not 'selected' but detections_list was supplied.", file=sys.stderr)
error_free = False
@@ -355,4 +343,4 @@ def validate(configuration: dict, skip_password_accessibility_check:bool=True) -
except Exception as e:
print("There was an error validation the configuration: [%s]" % (
str(e)), file=sys.stderr)
return None, setup_schema
return None, setup_schema
bin/docker_detection_tester/new_arguments.py
-106
View File
@@ -1,106 +0,0 @@
import argparse
import json
from modules import validate_args
import sys
DEFAULT_CONFIG_FILE = "defaults.json"
def main(args):
try:
with open(DEFAULT_CONFIG_FILE, 'r') as settings_file:
default_settings = json.load(settings_file)
except Exception as e:
print("Error loading settings file %s: %s"%(DEFAULT_CONFIG_FILE, str(e)), file=sys.stderr)
sys.exit(1)
parser = argparse.ArgumentParser(
description="Use 'SOME_PROGRAM_NAME_STRING --help' to get help with the arguments")
actions_parser = parser.add_subparsers(title="test action")
configure_parser = actions_parser.add_parser(
"configure", help="configure a test run")
configure_parser.add_argument(
'-o', '--output_config', required=True, help="Name of config file to generate")
test_parser = actions_parser.add_parser("test", help="run a test")
test_parser.add_argument('-b', '--branch', required=True,
help="The branch whose detections you would like to test. "\
"In order to calculate new/changed detections, the detections "\
"in this branch will be diffed against those in the 'develop' branch")
test_parser.add_argument(
'-pr', '--pull_request_number', required=False, help="Pull request number.")
VALID_DETECTION_TYPES = ['endpoint', 'cloud', 'network']
#Common Test Arguments
test_parser.add_argument('-t', '--types', type=str, action="append",
help="Detection types to test. Can be one or more of %s"%(VALID_DETECTION_TYPES))
test_parser.add_argument('-e', '--escu_package', type=argparse.FileType('rb'), required=False,
help="A previously generated ESCU PAcklage to use. If you pass this "\
"argument, a new ESCU package will not be generated. Note that this "\
"may cause newly-written detections to fail (for example, if they "\
"leverage macros that have been added or modified).")
test_parser.add_argument('-p','--persist_security_content', required=False, action="store_true",
help="Assumes security_content directory already exists. Don't check it out and overwrite it again. Saves "\
"time and allows you to test a detection that you've updated. Runs generate again in case you have "\
"updated macros or anything else. Especially useful for quick, local, iterative testing.")
test_parser.add_argument('-tag', '--container_tag', required=False, default = default_args['container_tag'],
help="The tag of the Splunk Container to use. Tags are located "\
"at https://hub.docker.com/r/splunk/splunk/tags")
test_parser.add_argument("-show", "--show_password", required=False, default=False, action='store_true',
help="Show the generated password to use to login to splunk. For a CI/CD run, "\
"you probably don't want this.")
test_parser.add_argument('-r','--reuse_image', required=False, default=True, action='store_true',
help="Should existing images be re-used, or should they be redownloaded?")
test_parser.add_argument('-i', '--interactive_failure', required=False, default=False, action='store_true',
help="If a test fails, should we pause before removing data so that the search can be debugged?")
#Mode settings
mode_parser = test_parser.add_subparsers(title="Test Modes", required=True)
#NEW
new_parser = mode_parser.add_parser("changes",
help="Test only the new or changed detections")
#SELECTED
selected_parser = mode_parser.add_parser("selected", help="Test only the detections from the target branch that "\
" are passed on the command line. These can be given as "\
"a list of files or as a file containing a list of files.")
selected_group = selected_parser.add_mutually_exclusive_group(required=True)
selected_group.add_argument('-df', '--detections_file', type=argparse.FileType('r'),
required=False, help="A file containing a list of detections to run, one per line")
selected_group.add_argument('-dl', '--detections_list',
required=False, help="The names of files that you want to test, separated by commas. "\
"Do not include spaces between the detections!")
#ALL
all_parser = mode_parser.add_parser("all",
help="Test all of the detections in the target branch. "\
"Note that this could take a very long time.")
args = parser.parse_args()
try:
validate_args.validate(args.__dict__)
except Exception as e:
print("Error validating command line arguments: [%s]"%(str(e)))
sys.exit(1)
if __name__ == "__main__":
main(sys.argv[1:])
bin/docker_detection_tester/test_config_github_actions.json
-1
View File
@@ -79,7 +79,6 @@
"branch": "BRANCH_DOES_NOT_EXIST_USE_CLI_ARGUMENT",
"commit_hash": null,
"container_tag": "latest",
"detections_file": null,
"detections_list": null,
"folders": [
"endpoint",
tests/cloud/high_number_of_login_failures_from_a_single_source.test.yml → tests/experimental/cloud/high_number_of_login_failures_from_a_single_source.test.yml
View File
tests/endpoint/print_processor_registry_autostart.test.yml → tests/experimental/endpoint/print_processor_registry_autostart.test.yml
View File