diff --git a/bin/contentctl_project/contentctl_core/domain/entities/detection.py b/bin/contentctl_project/contentctl_core/domain/entities/detection.py index 104cc3c772..9a41f26121 100644 --- a/bin/contentctl_project/contentctl_core/domain/entities/detection.py +++ b/bin/contentctl_project/contentctl_core/domain/entities/detection.py @@ -8,24 +8,34 @@ from datetime import datetime from typing import Union - -from bin.contentctl_project.contentctl_core.domain.entities.security_content_object import SecurityContentObject -from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import AnalyticsType +from bin.contentctl_project.contentctl_core.domain.entities.security_content_object import ( + SecurityContentObject, +) +from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import ( + AnalyticsType, +) from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import DataModel -from bin.contentctl_project.contentctl_core.domain.entities.detection_tags import DetectionTags +from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import ( + DetectionStatus, +) + + +from bin.contentctl_project.contentctl_core.domain.entities.detection_tags import ( + DetectionTags, +) from bin.contentctl_project.contentctl_core.domain.entities.deployment import Deployment from bin.contentctl_project.contentctl_core.domain.entities.unit_test import UnitTest from bin.contentctl_project.contentctl_core.domain.entities.macro import Macro from bin.contentctl_project.contentctl_core.domain.entities.lookup import Lookup from bin.contentctl_project.contentctl_core.domain.entities.baseline import Baseline from bin.contentctl_project.contentctl_core.domain.entities.playbook import Playbook -from bin.contentctl_project.contentctl_core.domain.entities.link_validator import LinkValidator +from bin.contentctl_project.contentctl_core.domain.entities.link_validator import ( + LinkValidator, +) from bin.contentctl_project.contentctl_core.domain.entities.deployment import Deployment import sys - - class Detection(BaseModel, SecurityContentObject): # detection spec name: str @@ -34,9 +44,9 @@ class Detection(BaseModel, SecurityContentObject): date: str author: str type: str - status: str + status: DetectionStatus description: str - data_source : list[str] + data_source: list[str] search: Union[str, dict] how_to_implement: str known_false_positives: str @@ -64,49 +74,50 @@ class Detection(BaseModel, SecurityContentObject): nes_fields: str = None providing_technologies: list = None - - # @validator('name') - # def name_max_length(cls, v, values): + # @validator('name')v + # def name_max_length(cls, v, values): # if len(v) > 67: # raise ValueError('name is longer then 67 chars: ' + v) # return v - @validator('name') + class Config: + use_enum_values = True + + @validator("name") def name_invalid_chars(cls, v): invalidChars = set(string.punctuation.replace("-", "")) if any(char in invalidChars for char in v): - raise ValueError('invalid chars used in name: ' + v) + raise ValueError("invalid chars used in name: " + v) return v - @validator('id') + @validator("id") def id_check(cls, v, values): try: uuid.UUID(str(v)) except: - raise ValueError('uuid is not valid: ' + values["name"]) + raise ValueError("uuid is not valid: " + values["name"]) return v - @validator('date') + @validator("date") def date_valid(cls, v, values): try: datetime.strptime(v, "%Y-%m-%d") except: - raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"]) + raise ValueError("date is not in format YYYY-MM-DD: " + values["name"]) return v - @validator('type') + @validator("type") def type_valid(cls, v, values): if v.lower() not in [el.name.lower() for el in AnalyticsType]: - raise ValueError('not valid analytics type: ' + values["name"]) + raise ValueError("not valid analytics type: " + values["name"]) return v - - @validator('description', 'how_to_implement') + @validator("description", "how_to_implement") def encode_error(cls, v, values, field): try: - v.encode('ascii') + v.encode("ascii") except UnicodeEncodeError: - raise ValueError('encoding error in ' + field.name + ': ' + values["name"]) + raise ValueError("encoding error in " + field.name + ": " + values["name"]) return v # @root_validator @@ -122,23 +133,36 @@ class Detection(BaseModel, SecurityContentObject): @root_validator def name_max_length(cls, values): # Check max length only for ESCU searches, SSA does not have that constraint - if 'ssa_' not in values['file_path']: + if "ssa_" not in values["file_path"]: if len(values["name"]) > 67: - raise ValueError('name is longer then 67 chars: ' + values["name"]) + raise ValueError("name is longer then 67 chars: " + values["name"]) return values - # @validator('references') # def references_check(cls, v, values): # return LinkValidator.SecurityContentObject_validate_references(v, values) - @validator('search') + @validator("search") def search_validate(cls, v, values): # write search validator return v - @validator('tests') + @validator("tests") def tests_validate(cls, v, values): - if values["status"] != "experimental" and not v: - raise ValueError('tests value is needed for: ' + values["name"]) - return v \ No newline at end of file + if values["status"] != DetectionStatus.production and not v: + raise ValueError( + "tests value is needed for production detection: " + values["name"] + ) + return v + + @validator("experimental", always=True) + def experimental_validate(cls, v, values): + if DetectionStatus(values["status"]) == DetectionStatus.experimental: + return True + return False + + @validator("deprecated", always=True) + def deprecated_validate(cls, v, values): + if DetectionStatus(values["status"]) == DetectionStatus.deprecated: + return True + return False diff --git a/bin/contentctl_project/contentctl_core/domain/entities/enums/enums.py b/bin/contentctl_project/contentctl_core/domain/entities/enums/enums.py index 345caf08a3..5d4ac3de10 100644 --- a/bin/contentctl_project/contentctl_core/domain/entities/enums/enums.py +++ b/bin/contentctl_project/contentctl_core/domain/entities/enums/enums.py @@ -7,6 +7,7 @@ class AnalyticsType(enum.Enum): hunting = 3 correlation = 4 + class DataModel(enum.Enum): Endpoint = 1 Network_Traffic = 2 @@ -26,6 +27,7 @@ class DataModel(enum.Enum): Risk = 16 Splunk_Audit = 17 + class SecurityContentType(enum.Enum): detections = 1 baselines = 2 @@ -38,14 +40,22 @@ class SecurityContentType(enum.Enum): unit_tests = 9 attack_data = 10 + class SecurityContentProduct(enum.Enum): ESCU = 1 SSA = 2 API = 3 CUSTOM = 4 + class SigmaConverterTarget(enum.Enum): CIM = 1 RAW = 2 OCSF = 3 - ALL = 4 \ No newline at end of file + ALL = 4 + + +class DetectionStatus(enum.Enum): + production = "production" + deprecated = "deprecated" + experimental = "experimental" diff --git a/bin/docker_detection_tester/modules/github_service.py b/bin/docker_detection_tester/modules/github_service.py index 9c17227081..e8a9db4820 100644 --- a/bin/docker_detection_tester/modules/github_service.py +++ b/bin/docker_detection_tester/modules/github_service.py @@ -21,127 +21,164 @@ 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_ROOT_PATH = "security_content/detections" +TEST_ROOT_PATH = "security_content/tests" DETECTION_FILE_EXTENSION = ".yml" -TEST_FILE_EXTENSION = ".test.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: Union[int,None] = 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: print("Getting handle on existing security_content repo!") self.security_content_repo_obj = git.Repo("security_content") else: - print("Checking out security_content repo!") + print("Checking out security_content repo!") self.security_content_repo_obj = self.clone_project( - SECURITY_CONTENT_URL, f"security_content", f"develop") + SECURITY_CONTENT_URL, f"security_content", f"develop" + ) - #Ensure that the branch name is valid - #Get all the branch names, prefixed with "origin/" - branch_names = [branch.name for branch in self.security_content_repo_obj.remote().refs] + # Ensure that the branch name is valid + # Get all the branch names, prefixed with "origin/" + branch_names = [ + branch.name for branch in self.security_content_repo_obj.remote().refs + ] - if "origin/%s"%(security_content_branch) not in branch_names: - raise(Exception("Branch name [%s] not found in valid branches. Try running \n"\ - "'git branch -a' to examine [%d] branches"%(security_content_branch, len(branch_names)))) + if "origin/%s" % (security_content_branch) not in branch_names: + raise ( + Exception( + "Branch name [%s] not found in valid branches. Try running \n" + "'git branch -a' to examine [%d] branches" + % (security_content_branch, len(branch_names)) + ) + ) - - if commit_hash is not None and PR_number is not None: - print(f"\n************\nWARNING - both the PR_number {PR_number} and the commit_hash {commit_hash} were provided. " - f"You should only pass neither or one of these. We will ASSUME you want to use the PR_number, not the commit_hash. " - f"Removing the commit_hash...\n************\n") + if commit_hash is not None and PR_number is not None: + print( + f"\n************\nWARNING - both the PR_number {PR_number} and the commit_hash {commit_hash} were provided. " + f"You should only pass neither or one of these. We will ASSUME you want to use the PR_number, not the commit_hash. " + f"Removing the commit_hash...\n************\n" + ) commit_hash = None - - if PR_number: - ret = subprocess.run(["git", "-C", "security_content/", "fetch", "origin", - "refs/pull/%d/head:%s" % (PR_number, security_content_branch)], capture_output=True) - #ret = subprocess.call(["git", "-C", "security_content/", "fetch", "origin", + ret = subprocess.run( + [ + "git", + "-C", + "security_content/", + "fetch", + "origin", + "refs/pull/%d/head:%s" % (PR_number, security_content_branch), + ], + capture_output=True, + ) + # ret = subprocess.call(["git", "-C", "security_content/", "fetch", "origin", # "refs/pull/%d/head:%s" % (PR_number, security_content_branch)]) - - + if ret.returncode != 0: - raise(Exception("Error checking out repository: [%s]"%(ret.stdout.decode("utf-8") + "\n" + ret.stderr.decode("utf-8")))) - + raise ( + Exception( + "Error checking out repository: [%s]" + % ( + ret.stdout.decode("utf-8") + + "\n" + + ret.stderr.decode("utf-8") + ) + ) + ) # No checking to see if the hash is to a commit inside of the branch - the user # has to do that by hand. - + # -- ensures that we check out the appropriate branch or commit hash. # Without --, there can be ambiguity if a file/folder exists with the # same name as the branch, causing the checkout to fail with error if commit_hash is not None: print("Checking out commit hash: [%s]" % (commit_hash)) - self.security_content_repo_obj.git.checkout(commit_hash, '--') + self.security_content_repo_obj.git.checkout(commit_hash, "--") else: - #Even if we have fetched a PR, we still MUST check out the branch to + # Even if we have fetched a PR, we still MUST check out the branch to # be able to do anything with it. Otherwise we won't have the files - print("Checking out branch: [%s]..." % - (security_content_branch), end='') + print("Checking out branch: [%s]..." % (security_content_branch), end="") sys.stdout.flush() - self.security_content_repo_obj.git.checkout( - security_content_branch, '--') + self.security_content_repo_obj.git.checkout(security_content_branch, "--") commit_hash = self.security_content_repo_obj.head.object.hexsha print("commit_hash %s" % (commit_hash)) self.commit_hash = commit_hash - + def update_and_commit_passed_tests(self, results: list[dict]) -> bool: - - def update_and_commit_passed_tests(self, results:list[dict])->bool: - changed_file_paths = [] for result in results: - detection_obj_path = os.path.join("security_content","detections",result['detection_file']) - + detection_obj_path = os.path.join( + "security_content", "detections", result["detection_file"] + ) + test_obj_path = detection_obj_path.replace("detections", "tests", 1) - test_obj_path = test_obj_path.replace(".yml",".test.yml") + test_obj_path = test_obj_path.replace(".yml", ".test.yml") detection_obj = testing_service.load_file(detection_obj_path) test_obj = testing_service.load_file(test_obj_path) - detection_obj['tags']['automated_detection_testing'] = 'passed' - #detection_obj['tags']['automated_detection_testing_date'] = datetime.datetime.today().strftime('%Y-%m-%d-%H:%M:%S') - - for o in test_obj['tests']: - if 'attack_data' in o: + detection_obj["tags"]["automated_detection_testing"] = "passed" + # detection_obj['tags']['automated_detection_testing_date'] = datetime.datetime.today().strftime('%Y-%m-%d-%H:%M:%S') + + for o in test_obj["tests"]: + if "attack_data" in o: datasets = [] - for dataset in o['attack_data']: - datasets.append(dataset['data']) - detection_obj['tags']['dataset'] = datasets + for dataset in o["attack_data"]: + datasets.append(dataset["data"]) + detection_obj["tags"]["dataset"] = datasets with open(detection_obj_path, "w") as f: yaml.dump(detection_obj, f, sort_keys=False, allow_unicode=True) - + changed_file_paths.append(detection_obj_path) - - relpaths = [pathlib.Path(*pathlib.Path(p).parts[1:]).as_posix() for p in changed_file_paths] - newpath = relpaths[0]+'.wow' + + relpaths = [ + pathlib.Path(*pathlib.Path(p).parts[1:]).as_posix() + for p in changed_file_paths + ] + newpath = relpaths[0] + ".wow" relpaths.append(newpath) - with open('security_content/' + newpath,'w') as d: - d.write("fake file") + with open("security_content/" + newpath, "w") as d: + d.write("fake file") print("status results:") - print(self.security_content_repo_obj.index.diff(self.security_content_repo_obj.head.commit)) - + print( + self.security_content_repo_obj.index.diff( + self.security_content_repo_obj.head.commit + ) + ) + if len(relpaths) > 0: - print('there is at least one changed file') - print(relpaths) + print("there is at least one changed file") + print(relpaths) self.security_content_repo_obj.index.add(relpaths) print("status results after add:") - print(self.security_content_repo_obj.index.diff(self.security_content_repo_obj.head.commit)) + print( + self.security_content_repo_obj.index.diff( + self.security_content_repo_obj.head.commit + ) + ) - commit_message = "The following detections passed detection testing. Their YAMLs have been updated and their datasets linked:\n - %s"%("\n - ".join(relpaths)) + commit_message = ( + "The following detections passed detection testing. Their YAMLs have been updated and their datasets linked:\n - %s" + % ("\n - ".join(relpaths)) + ) self.security_content_repo_obj.index.commit(commit_message) return True else: return False - - - return True def clone_project(self, url, project, branch): @@ -149,124 +186,177 @@ class GithubService: repo_obj = git.Repo.clone_from(url, project, branch=branch) return repo_obj + def prune_detections( + self, + detection_files: list[str], + types_to_test: list[str], + exclude_ssa: bool = True, + ) -> list[str]: - def prune_detections(self, - detection_files: list[str], - types_to_test: list[str], - exclude_ssa: bool = True) -> list[str]: - - - pruned_tests = [] - + for detection in detection_files: - + if os.path.basename(detection).startswith(SSA_PREFIX) and exclude_ssa: continue with open(detection, "r") as d: - description:dict = yaml.safe_load(d) - - + description: dict = yaml.safe_load(d) - detection_filepath_without_security_content = str( - pathlib.Path(*pathlib.Path(detection).parts[1:])) + pathlib.Path(*pathlib.Path(detection).parts[1:]) + ) # If no types are provided, then we will get everything - if description.get('type', None) in types_to_test and description.get('status', None) == "production": + if ( + description.get("type", None) in types_to_test + and description.get("status", None) == "production" + and not ( + description.get("tags", False) + and description["tags"].get("manual_test", False) + ) + ): - if len(description.get("tests",[])) == 0: - print(Exception(f"Detection {detection_filepath_without_security_content} has no tests/test section defined. Detection must include at least one test.")) - continue - raise(Exception(f"Detection {detection_filepath_without_security_content} has no tests/test section defined. Detection must include at least one test.")) + if len(description.get("tests", [])) == 0: + print( + Exception( + f"Detection {detection_filepath_without_security_content} has no tests/test section defined. Detection must include at least one test." + ) + ) + continue + raise ( + Exception( + f"Detection {detection_filepath_without_security_content} has no tests/test section defined. Detection must include at least one test." + ) + ) pruned_tests.append(detection_filepath_without_security_content) - + else: + print( + f"Ignore {detection} with status: {description.get('status', None)}" + ) # 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): + + # 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 - def ensure_paired_detection_and_test_files(self, detection_files: list[str], test_files: list[str], exclude_ssa: bool = True)->bool: - ''' + 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}"\ - + 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 + # 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) + 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") + 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)) + 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 + + # 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) + 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") + 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)) + 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: + + 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}" - + + 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}]" + 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 : + 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}" + 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}]" + 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) + def get_detection_files( + self, + mode: str, + folders: list[str], + types: list[str], + detections_list: Union[list[str], None], + ) -> list[str]: - def get_detection_files(self, mode: str, folders: list[str], types: list[str], - detections_list: Union[list[str], None]) -> list[str]: - - - if mode == "changes": tests = self.get_changed_detection_files(folders, types) elif mode == "selected": @@ -274,7 +364,10 @@ class GithubService: # 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: @@ -282,136 +375,131 @@ class GithubService: else: # impossible to get here print( - "Impossible to get here. Just kept to make the if/elif more self describing", file=sys.stderr) + "Impossible to get here. Just kept to make the if/elif more self describing", + file=sys.stderr, + ) sys.exit(1) elif mode == "all": - tests = self.get_all_tests_and_detections(folders, types) + tests = self.get_all_tests_and_detections(folders, types) else: print( - "Error, unsupported mode [%s]. Mode must be one of %s", file=sys.stderr) + "Error, unsupported mode [%s]. Mode must be one of %s", file=sys.stderr + ) sys.exit(1) - # filter out and detections that have the - # experimental: true - # or - # manual_test: Description here - tests_without_experimental_or_manual_test = [] - removed_experimental = 0 - removed_manual_test = 0 - REMOVE_EXPERIMENTAL = True - for test in tests: - test_obj = testing_service.load_file(test) - - if test_obj.get("experimental", False) or (not REMOVE_EXPERIMENTAL): - removed_experimental += 1 - - elif test_obj.get("tags", None) and test_obj["tags"].get("manual_test", False): - removed_manual_test += 1 - - else: - tests_without_experimental_or_manual_test.append(test) - - - print(f"Experimental tests removed from test set: {removed_experimental}") - print(f" Manual tests removed from test set: {removed_manual_test}") - return tests_without_experimental_or_manual_test + return tests - def get_selected_test_files(self, - detection_file_list: list[str], - types_to_test: list[str] = [ - "Anomaly", "Hunting", "TTP"]) -> list[str]: + def get_selected_test_files( + self, + detection_file_list: list[str], + types_to_test: list[str] = ["Anomaly", "Hunting", "TTP"], + ) -> list[str]: return self.prune_detections(detection_file_list, types_to_test) - - - - def get_all_tests_and_detections(self, - folders: list[str] = [ - 'endpoint', 'cloud', 'network'], - types_to_test: list[str] = [ - "Anomaly", "Hunting", "TTP"]) -> list[str]: + def get_all_tests_and_detections( + self, + folders: list[str] = ["endpoint", "cloud", "network"], + types_to_test: list[str] = ["Anomaly", "Hunting", "TTP"], + ) -> list[str]: detections = [] for folder in folders: - detections.extend(self.get_all_files_in_folder(os.path.join(DETECTION_ROOT_PATH, folder), "*")) + 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) - + 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_detection_files(self, folders=['endpoint', 'cloud', 'network'], types_to_test=["Anomaly", "Hunting", "TTP"]) -> list[str]: + def get_changed_detection_files( + self, + folders=["endpoint", "cloud", "network"], + types_to_test=["Anomaly", "Hunting", "TTP"], + ) -> list[str]: branch1 = self.security_content_branch - branch2 = 'develop' - g = git.Git('security_content') + branch2 = "develop" + g = git.Git("security_content") all_changed_test_files = [] all_changed_detection_files = [] - if branch1 != 'develop': + if branch1 != "develop": if self.commit_hash is None: - differ = g.diff('--name-status', branch2 + '...' + branch1) + differ = g.diff("--name-status", branch2 + "..." + branch1) else: - differ = g.diff('--name-status', branch2 + - '...' + self.commit_hash) + differ = g.diff("--name-status", branch2 + "..." + self.commit_hash) changed_files = differ.splitlines() for file_path in changed_files: # added or changed test files - if file_path.startswith('A') or file_path.startswith('M'): + if file_path.startswith("A") or file_path.startswith("M"): # changed detections - if 'detections' in file_path and os.path.basename(file_path).endswith('.yml'): + if "detections" in file_path and os.path.basename( + file_path + ).endswith(".yml"): all_changed_detection_files.append(file_path) else: - print("Looking for changed detections by diffing [%s] against [%s]. They are the same branch, so none were returned." % ( - branch1, branch2), file=sys.stderr) + print( + "Looking for changed detections by diffing [%s] against [%s]. They are the same branch, so none were returned." + % (branch1, branch2), + file=sys.stderr, + ) return [] - - all_changed_detection_files = [os.path.join("security_content", name.split( - '\t')[1]) for name in all_changed_detection_files if len(name.split('\t')) == 2] - + all_changed_detection_files = [ + os.path.join("security_content", name.split("\t")[1]) + for name in all_changed_detection_files + if len(name.split("\t")) == 2 + ] - #Trim out any of the tests/detection that are not in the selected folders, but at least print a notice + # Trim out any of the tests/detection that are not in the selected folders, but at least print a notice # to the user. - - changed_detection_files = [x for x in all_changed_detection_files if - (len(pathlib.Path(x).parts) > 3 and pathlib.Path(x).parts[2] in folders) ] - - - for missing in set(changed_detection_files).symmetric_difference(all_changed_detection_files): - print("Ignoring modified detecton [%s] not in set of selected folders: %s"%(missing,folders)) - - # Convert the test files to the detection file equivalent. - # Note that some of these tests may be baselines and their associated + + changed_detection_files = [ + x + for x in all_changed_detection_files + if (len(pathlib.Path(x).parts) > 3 and pathlib.Path(x).parts[2] in folders) + ] + + for missing in set(changed_detection_files).symmetric_difference( + all_changed_detection_files + ): + print( + "Ignoring modified detecton [%s] not in set of selected folders: %s" + % (missing, folders) + ) + + # Convert the test files to the detection file equivalent. + # Note that some of these tests may be baselines and their associated # detection could be in experimental or not in the experimental folder converted_test_files = [] - #for test_filepath in changed_test_files: + # for test_filepath in changed_test_files: # detection_filename = str(pathlib.Path( # *pathlib.Path(test_filepath).parts[-2:])).replace("tests", "detections", 1) # converted_test_files.append(detection_filename) - - - - + return self.prune_detections(changed_detection_files, types_to_test) - #detections_to_test,_,_ = self.filter_test_types(changed_detection_files) + # detections_to_test,_,_ = self.filter_test_types(changed_detection_files) # for f in detections_to_test: # file_path_base = os.path.splitext(f)[0].replace('detections', 'tests') + '.test' # file_path_new = file_path_base + '.yml' # if file_path_new not in changed_test_files: # changed_test_files.append(file_path_new) - #print("Total things to test (test files and detection files changed): [%d]"%(len(changed_test_files))) + # print("Total things to test (test files and detection files changed): [%d]"%(len(changed_test_files))) # for l in changed_test_files: # print(l) # print(len(changed_test_files)) - #import time + # import time # time.sleep(5) def filter_test_types(self, test_files, test_types=["Anomaly", "Hunting", "TTP"]): @@ -422,23 +510,32 @@ class GithubService: try: with open(os.path.join("security_content", filename), "r") as fileData: yaml_dict = list(yaml.safe_load_all(fileData))[0] - if 'type' not in yaml_dict.keys(): + if "type" not in yaml_dict.keys(): print( - "Failed to find 'type' in the yaml for: [%s]" % (filename)) + "Failed to find 'type' in the yaml for: [%s]" % (filename) + ) error_files.append(filename) - if yaml_dict['type'] in test_types: + if yaml_dict["type"] in test_types: files_to_test.append(filename) else: files_not_to_test.append(filename) except Exception as e: - print("Error on trying to scan [%s]: [%s]" % ( - filename, str(e))) + print("Error on trying to scan [%s]: [%s]" % (filename, str(e))) error_files.append(filename) - print("***Detection Information***\n" - "\tTotal Files : %d" - "\tFiles to test : %d" - "\tFiles not to test : %d" - "\tError files : %d" % (len(test_files), len(files_to_test), len(files_not_to_test), len(error_files))) + print( + "***Detection Information***\n" + "\tTotal Files : %d" + "\tFiles to test : %d" + "\tFiles not to test : %d" + "\tError files : %d" + % ( + len(test_files), + len(files_to_test), + len(files_not_to_test), + len(error_files), + ) + ) import time + time.sleep(5) return files_to_test, files_not_to_test, error_files