Lots of testing changes.... lots of progress. Need to clean up and test entire set of detections properly with sleeps and diagnose how long sleep should be or if there is a way to block while input data is processed entirely.

2026-06-08 17:32:49 +00:00 · 2021-10-11 17:05:28 -07:00
parent 06a9c0f15e
commit 2f06facd23
5 changed files with 173 additions and 40 deletions
@@ -13,8 +13,14 @@ from modules import aws_service, testing_service
 import time
 import subprocess
 from datetime import datetime
-index_file_container_path = "/opt/splunk/etc/apps/search/"
+
+SPLUNK_CONTAINER_APPS_DIR = "/opt/splunk/etc/apps"
 index_file_local_path = "indexes.conf.tar"
+index_file_container_path = os.path.join(SPLUNK_CONTAINER_APPS_DIR, "search")
+
+datamodel_file_local_path = "datamodels.conf.tar"
+datamodel_file_container_path = os.path.join(SPLUNK_CONTAINER_APPS_DIR, "Splunk_SA_CIM")
+

 PASSWORD_LENGTH=20
 MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING=2
@@ -90,6 +96,8 @@ def main(args):
    parser.add_argument("-i", "--reuse_images", required=False, type=bool, default=False, help="Should existing images be re-used, or should they be redownloaded?")
    parser.add_argument("-c", "--reuse_containers", required=False, type=bool, default=False,  help="Should existing containers be re-used, or should they be rebuilt?")

+    parser.add_argument("-s", "--success", type=str, required=False, help="File that contains previously successful runs that we don't need to test")
+
    args = parser.parse_args()
    branch = args.branch
    uuid_test = args.uuid
@@ -97,6 +105,13 @@ def main(args):
    num_containers = args.num_containers
    reuse_containers = args.reuse_containers
    reuse_images = args.reuse_images
+    success_file = args.success_file
+
+    if success_file is not None:
+        with open(success_file, "r") as successes:
+            success_tests = [x.strip() for x in successes.readlines()]
+    else:
+        success_tests = []


    if num_containers < 1:
@@ -119,11 +134,18 @@ def main(args):
        #aws_service.dynamo_db_nothing_to_test(REGION, uuid_test, str(int(time.time())))
        sys.exit(0)
    print("The files to test: %s", str(test_files))
+    new_test_files = []
+    for f in test_files:
+        if f not in success_tests:
+            new_test_files.append(f)
+        else:
+            print("Already found [%s] in success file, not testing it again"%(f))
+    test_files = new_test_files
    
    #Go into the security content directory
    print("****GENERATE NEW CONTENT****")
    os.chdir("security_content")
-    commands = ["python3 -m venv .venv", "source .venv/bin/activate", "python3 -m pip install -r requirements.txt", "python3 contentctl.py --path . --verbose generate --product ESCU --output dist/escu"]
+    commands = ["python3 -m venv .venv", "source .venv/bin/activate", "python3 -m pip install wheel", "python3 -m pip install -r requirements.txt", "python3 contentctl.py --path . --verbose generate --product ESCU --output dist/escu"]
    ret = subprocess.run("; ".join(commands), shell=True, capture_output=False)
    if ret.returncode != 0:
        print("Error generating new content.  Exiting...")
@@ -167,7 +189,6 @@ def main(args):
        test_file_queue.put(filename)
    # print("***Test files enqueued")

-    testing_service.test_detection_wrapper(None, None, splunk_password, None, test_file_queue.get(), 0%1, uuid_test)
    
    # print("Getting docker client")
    client = docker.client.from_env()
@@ -310,16 +331,28 @@ def main(args):
    results_queue = queue.Queue()
    success_names_queue = queue.Queue()
    failure_names_queue = queue.Queue()
+    
+
+    
    for container_index in range(num_containers):
        container_name = "%s_%d"%(BASE_CONTAINER_NAME, container_index)
        
        web_port = BASE_CONTAINER_WEB_PORT  + container_index
        management_port = BASE_CONTAINER_MANAGEMENT_PORT + container_index

+        SPLUNK_COMMON_INFORMATION_MODEL = "https://splunkbase.splunk.com/app/1621/release/4.20.2/download"        
+        SPLUNK_SECURITY_ESSENTIALS = "https://splunkbase.splunk.com/app/3435/release/3.3.4/download"
+        #SPLUNK_ADD_ON_FOR_SYSMON = "https://splunkbase.splunk.com/app/5709/release/1.0.1/download"
+        SPLUNK_ADD_ON_FOR_SYSMON = "https://splunkbase.splunk.com/app/1914/release/10.6.2/download"
+        SYSMON_APP_FOR_SPLUNK = "https://splunkbase.splunk.com/app/3544/release/2.0.0/download"
+        SPLUNK_ES_CONTENT_UPDATE = "https://splunkbase.splunk.com/app/3449/release/3.29.0/download"
+
+        SPLUNK_APPS = [SPLUNK_COMMON_INFORMATION_MODEL, SPLUNK_SECURITY_ESSENTIALS, SPLUNK_ADD_ON_FOR_SYSMON, SYSMON_APP_FOR_SPLUNK, SPLUNK_ES_CONTENT_UPDATE]
+
        #docker run -p8089:8089 -p 8000:8000 -e "SPLUNK_START_ARGS=--accept-license" -e "SPLUNK_PASSWORD=123456qwertyQWERTY" -e "SPLUNK_APPS_URL=https://splunkbase.splunk.com/app/3435/release/3.3.4/download,https://splunkbase.splunk.com/app/5709/release/1.0.1/download,https://splunkbase.splunk.com/app/3449/release/3.29.0/download,https://splunkbase.splunk.com/app/1621/release/4.20.2/download" -e "SPLUNKBASE_USERNAME=ericmcginnistwo" -e "SPLUNKBASE_PASSWORD=splunkSecondAccount5@" -name splunktemplate splunk/splunk:latest 
        environment = {"SPLUNK_START_ARGS": "--accept-license",
                        "SPLUNK_PASSWORD"  : splunk_password, 
-                        "SPLUNK_APPS_URL"   : "https://splunkbase.splunk.com/app/3435/release/3.3.4/download,https://splunkbase.splunk.com/app/5709/release/1.0.1/download,https://splunkbase.splunk.com/app/3449/release/3.29.0/download,https://splunkbase.splunk.com/app/1621/release/4.20.2/download",
+                        "SPLUNK_APPS_URL"   : ','.join(SPLUNK_APPS),
                        "SPLUNKBASE_USERNAME" : "emcginnistwo",
                        "SPLUNKBASE_PASSWORD" : "splunkSecondAccount5@"
                        }
@@ -360,6 +393,10 @@ def main(args):

    print("DONE!")
    #read all the results out from the output queue
+    strtime = str(int(time.time()))
+    #write success and failure
+    success_output = open("success_%s"%(strtime), "w") 
+    failure_output = open("failure_%s"%(strtime), "w")
    try:
        while True:

@@ -391,6 +428,23 @@ def queue_status_thread(total_tests_count, testing_queue, results_queue, success
        else:
            time.sleep(10)

+def copy_file_to_container(localFilePath, remoteFilePath, containerName, sleepTimeSeconds=5):
+    successful_copy = False
+    #need to use the low level client to put a file onto a container
+    apiclient = docker.APIClient()
+    while not successful_copy:
+        try:
+            with open(localFilePath,"rb") as fileData:
+                #splunk will restart a few times will installation of apps takes place so it will reload its indexes...
+                apiclient.put_archive(container=containerName, path=remoteFilePath, data=fileData)
+                successful_copy=True
+        except Exception as e:
+            print("Failed copy of [%s] file to CONTAINER:[%s]...we will try again"%(localFilePath, containerName))
+            time.sleep(10)
+            successful_copy=False
+    print("Successfully copied [%s] to [%s] on [%s]"%(localFilePath, remoteFilePath, containerName))
+            
+
 def splunk_container_manager(testing_queue, container_name, splunk_ip, splunk_password, splunk_port, uuid_test, results_queue, success_names_queue, failure_names_queue):
    print("Starting the container [%s] after a sleep"%(container_name))
    #Is this going to be safe to use in different threads
@@ -405,25 +459,22 @@ def splunk_container_manager(testing_queue, container_name, splunk_ip, splunk_pa
    container = client.containers.get(container_name)
    print("Starting the container [%s]"%(container_name))

-    #need to use the low level client to put a file onto a container
-    apiclient = docker.APIClient()
+    

    container.start()
-    successful_copy = False
-    while not successful_copy:
-        try:
-            with open(index_file_local_path,"rb") as indexData:
-                #splunk will restart a few times will installation of apps takes place so it will reload its indexes...
-                time.sleep(10)
-                apiclient.put_archive(container=container_name, path=index_file_container_path, data=indexData)
-                successful_copy=True
-        except Exception as e:
-            print("Failed copy of index file to CONTAINER:[%s]...we will try again"%(container_name))
-            successful_copy=False
-            
+    print("Start copying files to container")
+    copy_file_to_container(index_file_local_path, index_file_container_path, container_name)
+    copy_file_to_container(datamodel_file_local_path, datamodel_file_container_path, container_name)
+    print("Finished copying files to container!")


    wait_for_splunk_ready(max_seconds=120)
+    from modules.splunk_sdk import enable_delete_for_admin
+    if not enable_delete_for_admin(splunk_ip, splunk_port, splunk_password):
+        print("COULD NOT ENABLE DELETE FOR [%s].... quitting"%(container_name))
+        sys.exit(0)
+    
+    print("Successfully enabled DELETE for [%s]"%(container_name))

    index=0
    try:
@@ -431,7 +482,6 @@ def splunk_container_manager(testing_queue, container_name, splunk_ip, splunk_pa
            #Try to get something from the queue
            detection_to_test = testing_queue.get(block=False)
            
-            
            #There is a detection to test
            print("Container [%s]--->[%s]"%(container_name, detection_to_test))
            try:
@@ -4,7 +4,7 @@ import os
 import logging
 import glob
 import subprocess
-
+import yaml

 # Logger
 logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
@@ -33,7 +33,7 @@ class GithubService:
        branch2 = 'develop'
        g = git.Git('security_content')
        changed_test_files = []
-
+        changed_detection_files = []
        if branch1 != 'develop':
            differ = g.diff('--name-status', branch2 + '...' + branch1)
            changed_files = differ.splitlines()
@@ -49,12 +49,67 @@ class GithubService:
                    # changed detections
                    if 'detections' in file_path:
                        if not os.path.basename(file_path).startswith('ssa') and os.path.basename(file_path).endswith('.yml'):
-                            file_path_base = os.path.splitext(file_path)[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)
+                            changed_detection_files.append(file_path)
+                            #file_path_base = os.path.splitext(file_path)[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)

+        #all files have the format A\tFILENAME or M\tFILENAME.  Get rid of those leading characters
+        changed_test_files = [name.split('\t')[1] for name in changed_test_files if len(name.split('\t')) == 2]
+        changed_detection_files = [name.split('\t')[1] for name in changed_detection_files if len(name.split('\t')) == 2]
+
+
+        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)))
+        #for l in changed_test_files:
+        #    print(l)
+        #print(len(changed_test_files))
+        import time
+        time.sleep(5)
        return changed_test_files

+    def filter_test_types(self, test_files, test_types = ["Anomaly", "Hunting", "TTP"]):
+        files_to_test = []
+        files_not_to_test = []
+        error_files = []
+        for filename in test_files:
+            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():
+                        print("Failed to find 'type' in the yaml for: [%s]"%(filename))
+                        error_files.append(filename)
+                    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)))
+                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)))
+        import time
+        time.sleep(5)
+        return files_to_test, files_not_to_test, error_files    
+
+
+
+                
+
+


@@ -5,6 +5,31 @@ import splunklib.client as client
 import splunklib.results as results
 import requests

+
+def enable_delete_for_admin(splunk_host, splunk_port, splunk_password):
+    try:
+        service = client.connect(
+            host=splunk_host,
+            port=splunk_port,
+            username='admin',
+            password=splunk_password
+        )
+    except Exception as e:
+        print("Unable to connect to Splunk instance: " + str(e))
+        return 1, {}
+
+    # search and replace \\ with \\\
+    # search = search.replace('\\','\\\\')
+    role = service.roles['admin']
+    try:
+        role.grant('delete_by_keyword')
+    except Exception as e:
+        print("Error - failed trying to grant 'can_delete' privs to admin: [%s]"%(str(e)))
+        return False
+    return True
+
+
+
 def test_baseline_search(splunk_host, splunk_port, splunk_password, search, pass_condition, baseline_name, baseline_file, earliest_time, latest_time):
    try:
        service = client.connect(
@@ -82,8 +107,8 @@ def test_detection_search(splunk_host, splunk_port, splunk_password, search, pas
    splunk_search = search + ' ' + pass_condition
    print("SEARCH:")
    print(splunk_search)
-    print("Sleep for 30 seconds")
-    sleep(30)
+    #print("Sleep for 30 seconds")
+    #sleep(30)
    try:
        job = service.jobs.create(splunk_search, **kwargs)
    except Exception as e:
@@ -109,8 +134,7 @@ def test_detection_search(splunk_host, splunk_port, splunk_password, search, pas


 def delete_attack_data(splunk_host, splunk_password, splunk_port):
-    #print("DO NOT DELETE ANYTHING!")
-    #return None
+    print("Deleting test data!")
    try:
        service = client.connect(
            host=splunk_host,
@@ -122,8 +146,8 @@ def delete_attack_data(splunk_host, splunk_password, splunk_port):
        print("Unable to connect to Splunk instance: " + str(e))
        return 1, {}

-    splunk_search = 'search index=test* | delete'
-
+    #splunk_search = 'search index=test* | delete'
+    splunk_search = 'search index=main | delete'
    kwargs = {"exec_mode": "blocking",
              "dispatch.earliest_time": "-1d",
              "dispatch.latest_time": "now"}
@@ -28,8 +28,9 @@ def test_detection_wrapper(container_name, splunk_ip, splunk_password, splunk_po
    result_test = test_detection(splunk_ip, splunk_port, container_name, splunk_password, test_file, test_index, uuid_test, uuid_var)
    

+    #enter = input("Run some tests from [%s] on [%s] - we don't delete until you hit enter :)"%(container_name, test_file))
    # delete test data
-    #splunk_sdk.delete_attack_data(splunk_ip, splunk_password, splunk_port)
+    splunk_sdk.delete_attack_data(splunk_ip, splunk_password, splunk_port)

    
    if result_test['detection_result']['error']:
@@ -42,7 +43,7 @@ def test_detection_wrapper(container_name, splunk_ip, splunk_password, splunk_po

 def test_detection(splunk_ip, splunk_port, container_name, splunk_password, test_file, test_index, uuid_test, uuid_var):
    try:
-        test_file_obj = load_file("security_content/" + test_file[2:])
+        test_file_obj = load_file(os.path.join("security_content/", test_file))
    except Exception as e:
        raise
        #print('Error: ' + str(e))
@@ -64,18 +65,21 @@ def test_detection(splunk_ip, splunk_port, container_name, splunk_password, test
    for attack_data in test_file_obj['tests'][0]['attack_data']:
        url = attack_data['data']
        r = requests.get(url, allow_redirects=True)
-        open(folder_name + '/' + attack_data['file_name'], 'wb').write(r.content)
-        print(folder_name + '/' + attack_data['file_name'])
+        target_file = os.path.join(folder_name, attack_data['file_name'])
+        with open(target_file, 'wb') as target:
+            target.write(r.content)
+        print(target_file)

        # Update timestamps before replay
        if 'update_timestamp' in attack_data:
            if attack_data['update_timestamp'] == True:
                data_manipulation = DataManipulation()
-                data_manipulation.manipulate_timestamp(folder_name + '/' + attack_data['file_name'], attack_data['sourcetype'], attack_data['source'])
-
-        replay_attack_dataset(container_name, splunk_password, folder_name, 'test' + str(test_index), attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
+                data_manipulation.manipulate_timestamp(target_file, attack_data['sourcetype'], attack_data['source'])
+        INDEX_TO_REPLAY_INTO = 'test' + str(test_index)
+        INDEX_TO_REPLAY_INTO = 'main'
+        replay_attack_dataset(container_name, splunk_password, folder_name, INDEX_TO_REPLAY_INTO, attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
    print("START SLEEP AFTER REPLAY")
-    time.sleep(30)
+    time.sleep(60)
    print("DONE SLEEP AFTER REPLAY")
    result_test = {}
    test = test_file_obj['tests'][0]
@@ -120,7 +124,7 @@ def test_detection(splunk_ip, splunk_port, container_name, splunk_password, test
 def load_file(file_path):
    try:
        print("Opening file path [%s]"%(file_path))
-        sys.exit(0)
+        
        with open(file_path, 'r', encoding="utf-8") as stream:
            try:
                file = list(yaml.safe_load_all(stream))[0]