Branch was auto-updated.

bin/ssa-end-to-end-testing/modules/github_service.py

@@ -2,8 +2,6 @@
 import git
 import os
 import logging
-from os import path
-import sys
 
 
 # Logger
@@ -25,7 +23,6 @@ class GithubService:
         repo_obj = git.Repo.clone_from(url, project, branch=branch)
         return repo_obj
 
-
     def get_changed_test_files_ssa(self):
         branch1 = self.security_content_branch
         branch2 = 'develop'
@@ -35,8 +32,6 @@ class GithubService:
 
         changed_ssa_test_files = []
 
-        #tests = self.read_security_content_test_files()
-
         for file_path in changed_files:
             # added or changed test files
             if file_path.startswith('tests'):
@@ -47,7 +42,10 @@ class GithubService:
             # changed detections
             if file_path.startswith('detections'):
                 if os.path.basename(file_path).startswith('ssa'):
-                    file_path_new = os.path.splitext(file_path)[0].replace('detections', 'tests') + '.test.yml'
+                    file_path_base = os.path.splitext(file_path)[0].replace('detections', 'tests') + '.test'
+                    file_path_new = file_path_base + '.yml'
+                    if not os.path.exists(file_path_new):
+                        file_path_new = file_path_base + '.yaml'
                     if file_path_new not in changed_ssa_test_files:
                         changed_ssa_test_files.append(file_path_new)
 

bin/ssa-end-to-end-testing/modules/spl/detection.spl

@@ -17,4 +17,5 @@
 | eval start_time = timestamp,
 end_time = timestamp,
 entities = mvappend(dest_device_id, dest_user_id),
-body = "TBD";
+body = "TBD"
+| into write_ssa_detected_events();

bin/ssa-end-to-end-testing/modules/spl/detection2.spl

@@ -17,4 +17,5 @@
 | eval start_time = timestamp,
 end_time = timestamp,
 entities = mvappend(dest_device_id, dest_user_id),
-body = "TBD";
+body = "TBD"
+| into write_ssa_detected_events();

bin/ssa-end-to-end-testing/modules/spl/firehose.spl

@@ -1 +1 @@
-| from read_splunk_firehose();
+    | from read_splunk_firehose();

bin/ssa-end-to-end-testing/modules/spl/troubleshoot.spl

@@ -1 +1 @@
-| from read_ssa_enriched_events();
+| from read_ssa_enriched_events() | into write_ssa_detected_events();

bin/ssa-end-to-end-testing/modules/test_ssa_detections.py

@@ -1,4 +1,3 @@
-
 import logging
 import os
 import time
@@ -6,8 +5,7 @@ import sys
 
 from http import HTTPStatus
 from modules.streams_service_api_helper import DSPApi
-from modules.utils import manipulate_spl, read_spl, read_data
-
+from modules.utils import check_source_sink, manipulate_spl, read_spl, read_data
 
 # Logger
 logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
@@ -22,6 +20,7 @@ MAX_EXECUTION_TIME_LIMIT = 600  # per detection test
 
 TEST_DATASET = 'windows-security_small.txt'
 
+
 class SSADetectionTesting:
 
     def __init__(self, env, tenant, header_token):
@@ -45,7 +44,7 @@ class SSADetectionTesting:
         ]
 
         test_results = []
-        for i in range(0,len(test_spls)):
+        for i in range(0, len(test_spls)):
             self.max_execution_time = MAX_EXECUTION_TIME_LIMIT
             test_result = self.ssa_detection_test(read_spl(file_path_spl, test_spls[i]), file_path_data, test_names[i])
             test_results.append(test_result.copy())
@@ -62,17 +61,16 @@ class SSADetectionTesting:
 
         return passed
 
-
     def test_ssa_detections(self, test_obj):
         LOGGER.info('Test SSA Detection: ' + test_obj["detection_obj"]["name"])
         self.max_execution_time = MAX_EXECUTION_TIME_LIMIT
         file_path_attack_data = os.path.join(os.path.dirname(__file__), "../", test_obj["attack_data_file_path"])
 
-        test_results = self.ssa_detection_test(test_obj["detection_obj"]["search"], file_path_attack_data, "SSA Smoke Test " + test_obj["test_obj"]["name"])
+        test_results = self.ssa_detection_test(test_obj["detection_obj"]["search"], file_path_attack_data,
+                                               "SSA Smoke Test " + test_obj["test_obj"]["name"])
 
         return test_results
 
-
     ## Helper Functions ##
 
     def update_execution_time(self, time_frame):
@@ -86,58 +84,45 @@ class SSADetectionTesting:
         time.sleep(time_in_s)
         return self.update_execution_time(time_in_s)
 
-    def check_result(self, condition, error_message):
-        try:
-            assert condition
-        except:
-            self.execution_passed = False
-            LOGGER.error(error_message)
-
-    def write_test_results(self, test_name):
-        if not self.execution_passed:
-            msg = f"Detection test failed for {test_name}"
-            LOGGER.error(msg)
-            self.test_results["msg"] = msg
-            self.test_results["result"] = False
-        else:
-            msg = f"Detection test successful for {test_name}"
-            LOGGER.info(msg)
-            self.test_results["msg"] = msg
-
-
     def ssa_detection_test_init(self):
         self.test_results["result"] = True
         self.test_results["msg"] = ""
         self.results_index = self.api.create_temp_index("mc")
-
+        self.created_pipelines = []
+        self.activated_pipelines = []
 
     def ssa_detection_test_main(self, spl, source, test_name):
         self.execution_passed = True
 
         self.wait_time(SLEEP_TIME_CREATE_INDEX)
 
+        check_ssa_spl = check_source_sink(spl)
         spl = manipulate_spl(self.api.env, spl, self.results_index)
-        self.check_result(spl is not None, "fail to manipulate spl file")
+        assert spl is not None, "fail to manipulate spl file"
 
         pipeline_id = self.api.create_pipeline_from_spl(spl)
-        self.check_result(pipeline_id is not None, "failed to create a pipeline")
+        assert pipeline_id is not None, "failed to create a pipeline"
 
         _pipeline_status = self.api.pipeline_status(pipeline_id)
-        self.check_result(_pipeline_status=="CREATED", f"Current status of pipeline {pipeline_id} should be CREATED")
+        assert _pipeline_status == "CREATED", f"Current status of pipeline {pipeline_id} should be CREATED"
+        self.created_pipelines.append(pipeline_id)
 
         response_body = self.api.activate_pipeline(pipeline_id)
-        self.check_result(response_body.get("activated")==pipeline_id, f"pipeline {pipeline_id} should be successfully activate.")
+        assert response_body.get("activated") == pipeline_id, f"pipeline {pipeline_id} should be successfully activate."
+        self.activated_pipelines.append(pipeline_id)
 
         self.wait_time(SLEEP_TIME_ACTIVATE_PIPELINE)
 
+        if not check_ssa_spl:
+            msg = f"Detection test successful for {test_name}"
+            LOGGER.warning(f"Test not completed. Detection seems deprecated, and will not send messages to SSA")
+            self.test_results["msg"] = msg
+            return self.test_results
+
         data = read_data(source)
         LOGGER.info("Sending (%d) events" % (len(data)))
 
-        if len(data) == 0:
-            LOGGER.warning("No events to send, skip to next test.")
-            self.execution_passed = False
-            self.write_test_results(test_name)
-            return self.test_results
+        assert len(data) > 0, "No events to send, skip to next test."
 
         for d in data:
             response_body = self.api.ingest_data(d)
@@ -151,8 +136,8 @@ class SSADetectionTesting:
             max_execution_time_reached = self.wait_time(WAIT_CYCLE)
             query = f"from indexes('{self.results_index['name']}') | search source!=\"Search Catalog\" "
             sid = self.api.submit_search_job(self.results_index['module'], query)
-            self.check_result(sid is not None, f"Failed to create a Search Job")
-            
+            assert sid is not None, f"Failed to create a Search Job"
+
             job_finished = False
             while not job_finished:
                 self.wait_time(WAIT_CYCLE)
@@ -162,33 +147,53 @@ class SSADetectionTesting:
             results = self.api.get_search_job_results(sid)
             search_results = (len(results) > 0)
             if not search_results:
-                LOGGER.info(f"Search didn't return any results. Retrying in {WAIT_CYCLE}s, max execution time left {self.max_execution_time}s")
-        
-        self.check_result(len(results) > 0, "Search job didn't return any results")
+                LOGGER.info(
+                    f"Search didn't return any results. Retrying in {WAIT_CYCLE}s, max execution time left {self.max_execution_time}s")
 
-        response, response_body = self.api.deactivate_pipeline(pipeline_id)
-        self.check_result(response.status_code == HTTPStatus.OK, f"The pipeline {pipeline_id} fails to deactivated.")
+        assert len(results) > 0, "Search job didn't return any results"
 
-        response = self.api.delete_pipeline(pipeline_id)
-        self.check_result(response.status_code == HTTPStatus.NO_CONTENT, f"Fail to delete pipeline {pipeline_id}.")
-
-        self.write_test_results(test_name)
+        msg = f"Detection test successful for {test_name}"
+        LOGGER.info(msg)
+        self.test_results["msg"] = msg
 
         return self.test_results
 
-
     def ssa_detection_test_teardown(self):
-        pass
-        self.api.delete_temp_index(self.results_index["id"])
-
+        """
+        Deactivate and deletes pipelines, deletes results indexes,
+        and when it fails it shows pipelines and result indexes that were not removed.
+        :return:
+        None
+        """
+        deactivate_pipeline = lambda p: self.api.deactivate_pipeline(p)[0].status_code == HTTPStatus.OK
+        delete_pipeline = lambda p: self.api.delete_pipeline(p).status_code == HTTPStatus.NO_CONTENT
+        delete_index = lambda p: self.api.delete_temp_index(p["id"]) == HTTPStatus.NO_CONTENT
+        self.activated_pipelines = [p for p in self.activated_pipelines if not deactivate_pipeline(p)]
+        self.created_pipelines = [p for p in self.created_pipelines if not delete_pipeline(p)]
+        if len(self.activated_pipelines) > 0 or len(self.created_pipelines) > 0 or not delete_index(self.results_index):
+            LOGGER.warning("Not all SCS resources fred up")
+            LOGGER.info(f"Created Pipelines: {','.join(self.created_pipelines)}")
+            LOGGER.info(f"Active Pipelines: {','.join(self.activated_pipelines)}")
+            LOGGER.info(f"Result Indexes: {self.results_index}")
+        else:
+            LOGGER.info("Testing successfully cleaned up")
 
     def ssa_detection_test(self, spl, source, test_name):
         self.ssa_detection_test_init()
-        test_result = self.ssa_detection_test_main(spl, source, test_name)
-        self.ssa_detection_test_teardown()
-        return test_result
-
-
+        try:
+            test_result = self.ssa_detection_test_main(spl, source, test_name)
+            self.ssa_detection_test_teardown()
+            return test_result
+        except AssertionError as e:
+            self.ssa_detection_test_teardown()
+            LOGGER.error(e.args[0])
+            return {"result": False,
+                    "msg": f"Detection test failure for {test_name}"}
+        except Exception as e:
+            self.ssa_detection_test_teardown()
+            LOGGER.error(e)
+            return {"result": False,
+                    "msg": f"Detection test failure for {test_name} (perhaps SCS problems)"}
 
     # only for troubleshooting
     # def ssa_detection_in_dsp_with_preview_session(self, spl, source, test_name):
@@ -213,4 +218,4 @@ class SSADetectionTesting:
 
     #     response = self.api.stop_preview_session(preview_id)
 
-    #     self.write_test_results(test_name)
+    #     self.write_test_results(test_name)

bin/ssa-end-to-end-testing/modules/utils.py

@@ -75,14 +75,26 @@ def request_headers(header_token):
     return headers
 
 
+def check_source_sink(spl):
+    match_source = re.match(r"^\s*\|\s+from\s+read_ssa_enriched_events\(\s*\)", spl)
+    match_sink = re.search(r"\|\s*into\s+write_ssa_detected_events\(\s*\)\s*;", spl)
+    return match_source and match_sink
+
+
 def manipulate_spl(env, spl, results_index):
-    spl = replace_ssa_macros(env, spl)
+    # Obtain the SSA source
+    pulsar_source_connection_id, pulsar_source_topic = return_macros(env)
+    source = READ_SSA_ENRICHED_EVENTS_EXPANDED\
+        .replace("__PULSAR_SOURCE_CONNECTION_ID__", pulsar_source_connection_id)\
+        .replace("__PULSAR_SOURCE_TOPIC__", pulsar_source_topic)
+    # Obtain the test sink
+    sink = ";"
     if results_index is not None:
-        # When an index is defined for a test, it writes the output of this pipeline to this index.
-        # original_pipeline; => original_pipeline | into index("module", "index");
         module = results_index["module"]
         index = results_index["name"]
-        spl = spl[:spl.rindex(";")] + f" | into index(\"{module}\", \"{index}\");"
+        sink = f"| into index(\"{module}\", \"{index}\");"
+    # Replace spl template with its `source` and `sink`
+    spl = replace_ssa_macros(source, sink, spl)
     LOGGER.info(f"spl: {spl}")
     return spl
 
@@ -93,13 +105,9 @@ def read_spl(file_path, file_name):
     return spl
 
 
-def replace_ssa_macros(env, spl):
-    pulsar_source_connection_id, pulsar_source_topic = return_macros(env)
-    macro_expanded = READ_SSA_ENRICHED_EVENTS_EXPANDED.replace("__PULSAR_SOURCE_CONNECTION_ID__", pulsar_source_connection_id)
-    macro_expanded = macro_expanded.replace("__PULSAR_SOURCE_TOPIC__", pulsar_source_topic)
-    spl = spl.replace(READ_SSA_ENRICHED_EVENTS, macro_expanded)
-    spl = spl.replace(WRITE_SSA_DETECTED_EVENTS, ";")
-    #spl = spl.replace("\n", " ")
+def replace_ssa_macros(source, sink, spl):
+    spl = spl.replace(READ_SSA_ENRICHED_EVENTS, source)
+    spl = spl.replace(WRITE_SSA_DETECTED_EVENTS, sink)
     return spl
 
 

bin/ssa-end-to-end-testing/run_ssa_smoketest.py

@@ -38,7 +38,7 @@ def main(args):
     for test_file in test_files_ssa:
         LOGGER.info(test_file)
 
-    if len(test_files_ssa)==0:
+    if len(test_files_ssa) == 0:
         LOGGER.info('Nothing to test for SSA smoke test.')
         sys.exit(0)
 

detections/endpoint/ssa___rare_parent_process_relationship_lolbas.yml

@@ -47,7 +47,7 @@ search: '| from read_ssa_enriched_events() | eval timestamp=parse_long(ucast(map
   OR process_name="pcalua.exe" OR process_name="cmdkey.exe" OR process_name="msconfig.exe")
 
   | eval start_time = timestamp, end_time = timestamp, entities = mvappend(dest_device_id,
-  dest_user_id), body = "TBD" | into write_ssa_detected_events();'
+  dest_user_id), body = "TBD" | into write_null();'
 how_to_implement: Collect endpoint data such as sysmon or 4688 events.
 known_false_positives: 'Some custom tools used by admins could be used rarely to launch
   remotely applications. This might trigger false positives at the beginning when

tests/endpoint/ssa___rare_parent_process_relationship_lolbas.test.yaml

@@ -2,7 +2,7 @@ name: Rare Parent/Child Process Relationship - SSA Unit Test
 tests:
   - name: Access LSASS Memory for Dump Creation
     file: endpoint/ssa___rare_parent_process_relationship_lolbas.yml
-    pass_condition: '@count_gt(0)'
+    pass_condition: '@count_eq(0)'
     description: Test detection looking for LOLBAS processes spawned by other processes that are rarely seen together
     attack_data:
       - file_name: T1059.all.labeled.lolbas-test.json