From d55a6cb9cd29bc8e115643b14d1d75e02a77c562 Mon Sep 17 00:00:00 2001 From: pyth0n1c <87383215+pyth0n1c@users.noreply.github.com> Date: Thu, 17 Feb 2022 17:47:15 -0800 Subject: [PATCH] Removed a lot of dead git repo code. Also allowed the branch --develop to be used when checking for changed files. This means that in local mode, local changes can be quickly and easily tested on the develop branch for smoke testing without the need to create a totally different branch. --- .../modules/github_service.py | 163 +++--------------- 1 file changed, 24 insertions(+), 139 deletions(-) diff --git a/bin/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py b/bin/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py index 1c3ff43327..b610cf7e54 100644 --- a/bin/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py +++ b/bin/automated_detection_testing/ci/detection_testing_batch/modules/github_service.py @@ -243,112 +243,6 @@ class GithubService: self.commit_hash = self.security_content_repo_obj.head.object.hexsha - - - - ''' - #If the directory exists, we do a checkout on develop and pull to make sure we have the latest content for develop - try: - if os.path.exists(SECURITY_CONTENT_DIRECTORY_NAME): - print(f"SECURITY_CONTENT_DIRECTORY_NAME exists. Getting the most updated version of the main branch ({SECURITY_CONTENT_MAIN_BRANCH})...",end='') - sys.stdout.flush() - - self.security_content_repo_obj = git.Repo(SECURITY_CONTENT_DIRECTORY_NAME) - - #Make sure we're on the right branch (develop) - self.security_content_repo_obj.git.checkout(SECURITY_CONTENT_MAIN_BRANCH) - - #Make sure that branch is up to date - self.security_content_repo_obj.git.pull() - - - - - else: - #the folder did not exist, so clone the repo - print(f"{SECURITY_CONTENT_DIRECTORY_NAME} does not exist. Downloading it and switching to the main branch ({SECURITY_CONTENT_MAIN_BRANCH})...",end='') - sys.stdout.flush() - - self.security_content_repo_obj = self.clone_project(SECURITY_CONTENT_URL, f"security_content", f"develop") - except Exception as e: - raise(Exception(f"Failure updating {SECURITY_CONTENT_DIRECTORY_NAME} folder - is it a valid repo? - [{str(e)}]")) - - print("done") - - if self.PR_number is not None: - self.security_content_remote_branch = security_content_branch - self.security_content_local_branch = LOCAL_PR_BRANCH%self.PR_number - else: - self.security_content_remote_branch = security_content_branch - self.security_content_local_branch = security_content_branch - - - #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] - - #Validate that the branch name is valid. For a PR, this should be the branch that the PR resides in - if "origin/%s"%(self.security_content_remote_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"%(self.security_content_remote_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") - commit_hash = None - - - - if PR_number: - print(f"Fetching PR {self.PR_number} into local branch {self.security_content_local_branch}...",end='') - self.security_content_repo_obj.git.fetch("origin", f"pull/{self.PR_number}/head:{self.security_content_local_branch}") - self.security_content_repo_obj.git.checkout(self.security_content_local_branch) - - elif commit_hash is not None: - # No checking to see if the hash is to a commit inside of the branch - the user - # has to do that by hand. - print("Checking out commit hash: [%s]..." % (commit_hash),end='') - self.security_content_repo_obj.git.checkout(commit_hash) - - else: - #Just checking out and testing a branch - print("Checking out branch: [%s]..." % - (security_content_branch), end='') - sys.stdout.flush() - self.security_content_repo_obj.git.checkout(security_content_branch) - #Pull to make sure we are up to date - - - #Print completion and store the commit hash - - print(f"done") - if (commit_hash != None) and (commit_hash != self.security_content_repo_obj.head.object.hexsha): - print("Warning: commit hash in configuration file [{commit_hash}] did NOT match the commit " - "hash of the PR [{self.security_content_repo_obj.head.object.hexsha}]. We just warn " - "about this and update the commit hash in the new test file.") - - commit_hash = self.security_content_repo_obj.head.object.hexsha - self.commit_hash = commit_hash - - if self.PR_number is not None: - #Verify that, if this was a PR, that the PR exists in the branch we believe it does - output = self.security_content_repo_obj.git.branch("-a", "--contains", commit_hash) - print(output) - print("***") - print(f"remotes/origin/{self.security_content_remote_branch}") - if f"remotes/origin/{self.security_content_remote_branch}" not in output: - #raise(Exception(f"PR number {self.PR_number} not found in branch {self.security_content_remote_branch}")) - print(f"PR number {self.PR_number} not found in branch {self.security_content_remote_branch}") - print("maybe this is a remote pr? or maybe you gave the wrong branch?") - - ''' - - - - - def update_and_commit_passed_tests(self, results:list[dict])->bool: changed_file_paths = [] @@ -554,31 +448,28 @@ class GithubService: all_changed_test_files = [] all_changed_detection_files = [] - if branch1 != SECURITY_CONTENT_MAIN_BRANCH: - #differ = g.diff('--name-status', branch2 + '...' + branch1) - - #diffs against the current state of the repo! - #this form is required if there have been local change or a commit hash. - #So just use it all the time - no drawbacks if just a branch is supplied - differ = g.diff('--name-status', branch2) + + #differ = g.diff('--name-status', branch2 + '...' + branch1) + #diffs against the current state of the repo! + #this form is required if there have been local change or a commit hash. + #So just use it all the time - no drawbacks if just a branch is supplied + differ = g.diff('--name-status', branch2) + - changed_files = differ.splitlines() + 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 'tests' in file_path and os.path.basename(file_path).endswith('.test.yml'): - all_changed_test_files.append(file_path) + for file_path in changed_files: + # added or changed test files + if file_path.startswith('A') or file_path.startswith('M'): + if 'tests' in file_path and os.path.basename(file_path).endswith('.test.yml'): + all_changed_test_files.append(file_path) - # changed detections - 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) - return [] + # changed detections + if 'detections' in file_path and os.path.basename(file_path).endswith('.yml'): + all_changed_detection_files.append(file_path) + # all files have the format A\tFILENAME or M\tFILENAME. Get rid of those leading characters @@ -636,21 +527,15 @@ class GithubService: changed_detection_files.append(name) - return self.prune_detections(changed_detection_files, types_to_test, previously_successful_tests) + pruned_detections = self.prune_detections(changed_detection_files, types_to_test, previously_successful_tests) - #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) + if len(pruned_detections) == 0 and branch1 == SECURITY_CONTENT_MAIN_BRANCH: + print(f"Looking for changed detections by diffing [{branch1}] against [{branch2}]. " + "They are the same branch and there have been no local changes, so no changed " + "test files were returned.", file=sys.stderr) + return pruned_detections - #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 - # time.sleep(5) + def filter_test_types(self, test_files, test_types=["Anomaly", "Hunting", "TTP"]): files_to_test = []