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

new detection testing

Browse Source
This commit is contained in:
P4T12ICK
2021-06-09 12:50:53 +02:00
parent 8fe77347ad
commit e7a144bd2d
3 changed files with 25 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py
+10 -10
View File
@@ -23,19 +23,19 @@ def main(args):
github_service = GithubService(branch)
test_files = github_service.get_changed_test_files()
for test_file in test_files:
print(test_file)
# for test_file in test_files:
# print(test_file)
# dt_ar = aws_service.get_ar_information_from_dynamo_db(REGION, DT_ATTACK_RANGE_STATE)
# splunk_instance = aws_service.get_splunk_instance(REGION, dt_ar['ssh_key_name'])
dt_ar = aws_service.get_ar_information_from_dynamo_db(REGION, DT_ATTACK_RANGE_STATE)
splunk_instance = aws_service.get_splunk_instance(REGION, dt_ar['ssh_key_name'])
# splunk_ip = splunk_instance['NetworkInterfaces'][0]['Association']['PublicIp']
# splunk_password = dt_ar['password']
# ssh_key_name = dt_ar['ssh_key_name']
# private_key = dt_ar['private_key']
splunk_ip = splunk_instance['NetworkInterfaces'][0]['Association']['PublicIp']
splunk_password = dt_ar['password']
ssh_key_name = dt_ar['ssh_key_name']
private_key = dt_ar['private_key']
# testing_service.prepare_detection_testing(ssh_key_name, private_key, splunk_ip, splunk_password)
# testing_service.test_detections(ssh_key_name, private_key, splunk_ip, splunk_password, test_files)
testing_service.prepare_detection_testing(ssh_key_name, private_key, splunk_ip, splunk_password)
testing_service.test_detections(ssh_key_name, private_key, splunk_ip, splunk_password, test_files)
# Get Password, private_key and key_name from DynamoDB
# For loop
automated_detection_testing/ci/detection_testing_batch/modules/github_service.py
-3
View File
@@ -19,7 +19,6 @@ class GithubService:
self.security_content_branch = security_content_branch
self.security_content_repo_obj = self.clone_project(SECURITY_CONTENT_URL, f"security_content", f"develop")
self.security_content_repo_obj.git.checkout(security_content_branch)
self.security_content_repo_obj.git.merge('develop')
def clone_project(self, url, project, branch):
LOGGER.info(f"Clone Security Content Project")
@@ -33,8 +32,6 @@ class GithubService:
changed_test_files = []
if branch1 != 'develop':
# differ = g.diff(branch1, branch2)
# print(differ)
differ = g.diff('--name-status', branch1, branch2)
changed_files = differ.splitlines()
automated_detection_testing/ci/detection_testing_batch/modules/testing_service.py
+15 -8
View File
@@ -30,8 +30,10 @@ def test_detections(ssh_key_name, private_key, splunk_ip, splunk_password, test_
def test_detection(ssh_key_name, private_key, splunk_ip, splunk_password, test_file, test_index):
test_file_obj = load_file("security_content/" + test_file)
#print(test_file_obj)
test_file_obj = load_file("security_content/" + test_file[2:])
if not test_file_obj:
return
print(test_file_obj)
epoch_time = str(int(time.time()))
folder_name = "attack_data_" + epoch_time
@@ -48,7 +50,7 @@ def test_detection(ssh_key_name, private_key, splunk_ip, splunk_password, test_f
data_manipulation = DataManipulation()
data_manipulation.manipulate_timestamp(folder_name + '/' + attack_data['file_name'], attack_data['sourcetype'], attack_data['source'])
replay_attack_dataset(splunk_ip, splunk_password, ssh_key_name, folder_name, 'test' + test_index, attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
replay_attack_dataset(splunk_ip, splunk_password, ssh_key_name, folder_name, 'test' + str(test_index), attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
if test_index == 10:
@@ -58,11 +60,16 @@ def test_detection(ssh_key_name, private_key, splunk_ip, splunk_password, test_f
def load_file(file_path):
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
sys.exit("ERROR: reading {0}".format(file_path))
try:
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
print("ERROR: reading {0}".format(file_path))
return False
except Exception as e:
print("ERROR: reading {0}".format(file_path))
return False
return file