Merge branch 'main' into feature/risk-observable-matching

Author: cmcginley-splunk

README.md

@@ -10,13 +10,13 @@ If you are already familiar with contentctl, the following common commands may b
 |-----------|---------|
 | Create a repository | `contentctl init` |
 | Validate Your Content | `contentctl validate` |
-| Validate Your Content, performing MITRE Enrichments | `contentctl validate –-enrichments`|
+| Validate Your Content, performing MITRE Enrichments | `contentctl validate --enrichments`|
 | Build Your App | `contentctl build` |
-| Test All the content in your app, pausing so that you can debug a search if it fails | `contentctl test –-post-test-behavior pause_on_failure mode:all` |
-| Test All the content in your app, pausing after every detection to allow debugging | `contentctl test –-post-test-behavior always_pause mode:all` |
-| Test 1 or more specified detections. If you are testing more than one detection, the paths are space-separated. You may also use shell-expanded regexes | `contentctl test –-post-test-behavior always_pause mode:selected --mode.files detections/endpoint/7zip_commandline_to_smb_share_path.yml detections/cloud/aws_multi_factor_authentication_disabled.yml detections/application/okta*` |
-| Diff your current branch with a target_branch and test detections that have been updated. Your current branch **must be DIFFERENT** than the target_branch | `contentctl test –-post-test-behavior always_pause mode:changes –-mode.target_branch develop` |
-| Perform Integration Testing of all content. Note that Enterprise Security MUST be listed as an app in your contentctl.yml folder, otherwise all tests will subsequently fail | `contentctl test –-enable-integration-testing --post-test-behavior never_pause mode:all` |
+| Test All the content in your app, pausing so that you can debug a search if it fails | `contentctl test --post-test-behavior pause_on_failure mode:all` |
+| Test All the content in your app, pausing after every detection to allow debugging | `contentctl test --post-test-behavior always_pause mode:all` |
+| Test 1 or more specified detections. If you are testing more than one detection, the paths are space-separated. You may also use shell-expanded regexes | `contentctl test --post-test-behavior always_pause mode:selected --mode.files detections/endpoint/7zip_commandline_to_smb_share_path.yml detections/cloud/aws_multi_factor_authentication_disabled.yml detections/application/okta*` |
+| Diff your current branch with a target_branch and test detections that have been updated. Your current branch **must be DIFFERENT** than the target_branch | `contentctl test --post-test-behavior always_pause mode:changes --mode.target_branch develop` |
+| Perform Integration Testing of all content. Note that Enterprise Security MUST be listed as an app in your contentctl.yml folder, otherwise all tests will subsequently fail | `contentctl test --enable-integration-testing --post-test-behavior never_pause mode:all` |
 
 # Introduction
 #### Security Is Hard 

contentctl/actions/build.py

@@ -8,7 +8,6 @@
 from contentctl.input.director import Director, DirectorOutputDto
 from contentctl.output.conf_output import ConfOutput
 from contentctl.output.conf_writer import ConfWriter
-from contentctl.output.ba_yml_output import BAYmlOutput
 from contentctl.output.api_json_output import ApiJsonOutput
 from contentctl.output.data_source_writer import DataSourceWriter
 from contentctl.objects.lookup import Lookup
@@ -86,17 +85,4 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:
 
             print(f"Build of '{input_dto.config.app.title}' API successful to {input_dto.config.getAPIPath()}")
 
-        if input_dto.config.build_ssa:
-            
-            srs_path = input_dto.config.getSSAPath() / 'srs'
-            complex_path = input_dto.config.getSSAPath() / 'complex'
-            shutil.rmtree(srs_path, ignore_errors=True)
-            shutil.rmtree(complex_path, ignore_errors=True)
-            srs_path.mkdir(parents=True)
-            complex_path.mkdir(parents=True)
-            ba_yml_output = BAYmlOutput()
-            ba_yml_output.writeObjects(input_dto.director_output_dto.ssa_detections, str(input_dto.config.getSSAPath()))
-
-            print(f"Build of 'SSA' successful to {input_dto.config.getSSAPath()}")
-                
         return input_dto.director_output_dto

contentctl/actions/convert.py

@@ -1054,15 +1054,20 @@ def retry_search_until_timeout(
             results = JSONResultsReader(job.results(output_mode="json"))
 
             # Consolidate a set of the distinct observable field names
-            observable_fields_set = set([o.name for o in detection.tags.observable])
+            observable_fields_set = set([o.name for o in detection.tags.observable]) # keeping this around for later
+            risk_object_fields_set = set([o.name for o in detection.tags.observable if "Victim" in o.role ]) # just the "Risk Objects"
+            threat_object_fields_set = set([o.name for o in detection.tags.observable if "Attacker" in o.role]) # just the "threat objects"
 
             # Ensure the search had at least one result
             if int(job.content.get("resultCount", "0")) > 0:
                 # Initialize the test result
                 test.result = UnitTestResult()
 
                 # Initialize the collection of fields that are empty that shouldn't be
+                present_threat_objects: set[str] = set()
                 empty_fields: set[str] = set()
+                
+                
 
                 # Filter out any messages in the results
                 for result in results:
@@ -1071,30 +1076,50 @@ def retry_search_until_timeout(
 
                     # If not a message, it is a dict and we will process it
                     results_fields_set = set(result.keys())
+                    # Guard against first events (relevant later)
 
-                    # Identify any observable fields that are not available in the results
-                    missing_fields = observable_fields_set - results_fields_set
-                    if len(missing_fields) > 0:
+                    # Identify any risk object fields that are not available in the results
+                    missing_risk_objects = risk_object_fields_set - results_fields_set
+                    if len(missing_risk_objects) > 0:
                         # Report a failure in such cases
-                        e = Exception(f"The observable field(s) {missing_fields} are missing in the detection results")
+                        e = Exception(f"The observable field(s) {missing_risk_objects} are missing in the detection results")
                         test.result.set_job_content(
                             job.content,
                             self.infrastructure,
-                            TestResultStatus.ERROR,
+                            TestResultStatus.FAIL,
                             exception=e,
                             duration=time.time() - search_start_time,
                         )
 
-                        return
+                        return                    
 
-                    # If we find one or more fields that contain the string "null" then they were
+                    # If we find one or more risk object fields that contain the string "null" then they were
                     # not populated and we should throw an error.  This can happen if there is a typo
                     # on a field.  In this case, the field will appear but will not contain any values
-                    current_empty_fields = set()
+                    current_empty_fields: set[str] = set()
+                    
                     for field in observable_fields_set:
                         if result.get(field, 'null') == 'null':
-                            current_empty_fields.add(field)
-
+                            if field in risk_object_fields_set:
+                                e = Exception(f"The risk object field {field} is missing in at least one result.")
+                                test.result.set_job_content(
+                                    job.content,
+                                    self.infrastructure,
+                                    TestResultStatus.FAIL,
+                                    exception=e,
+                                    duration=time.time() - search_start_time,
+                                )
+                                return
+                            else:
+                                if field in threat_object_fields_set:
+                                    current_empty_fields.add(field)
+                        else:
+                            if field in threat_object_fields_set:
+                                present_threat_objects.add(field)
+                                continue
+                                
+
+                    
                     # If everything succeeded up until now, and no empty fields are found in the
                     # current result, then the search was a success
                     if len(current_empty_fields) == 0:
@@ -1108,21 +1133,32 @@ def retry_search_until_timeout(
 
                     else:
                         empty_fields = empty_fields.union(current_empty_fields)
-
-                # Report a failure if there were empty fields in all results
-                e = Exception(
-                    f"One or more required observable fields {empty_fields} contained 'null' values.  Is the "
-                    "data being parsed correctly or is there an error in the naming of a field?"
+                        
+                        
+                missing_threat_objects = threat_object_fields_set - present_threat_objects
+                # Report a failure if there were empty fields in a threat object in all results
+                if len(missing_threat_objects) > 0:
+                    e = Exception(
+                        f"One or more required threat object fields {missing_threat_objects} contained 'null' values in all events. "
+                        "Is the data being parsed correctly or is there an error in the naming of a field?"
                     )
-                test.result.set_job_content(
-                    job.content,
-                    self.infrastructure,
-                    TestResultStatus.ERROR,
-                    exception=e,
-                    duration=time.time() - search_start_time,
-                )
+                    test.result.set_job_content(
+                        job.content,
+                        self.infrastructure,
+                        TestResultStatus.FAIL,
+                        exception=e,
+                        duration=time.time() - search_start_time,
+                    )
+                    return
+                
 
-                return
+                test.result.set_job_content(
+                            job.content,
+                            self.infrastructure,
+                            TestResultStatus.PASS,
+                            duration=time.time() - search_start_time,
+                        )
+                return               
 
             else:
                 # Report a failure if there were no results at all

contentctl/actions/detection_testing/infrastructures/DetectionTestingInfrastructure.py

@@ -30,7 +30,6 @@ def execute(self, input_dto: validate) -> DirectorOutputDto:
             [],
             [],
             [],
-            [],
         )
 
         director = Director(director_output_dto)

contentctl/actions/validate.py

@@ -7,7 +7,7 @@
 import logging
 from pydantic import BaseModel, Field
 from dataclasses import field
-from typing import Annotated
+from typing import Annotated,Any
 from contentctl.objects.mitre_attack_enrichment import MitreAttackEnrichment
 from contentctl.objects.config import validate
 logging.getLogger('taxii2client').setLevel(logging.CRITICAL)
@@ -33,21 +33,33 @@ def getEnrichmentByMitreID(self, mitre_id:Annotated[str, Field(pattern=r"^T\d{4}
         else:
             raise Exception(f"Error, Unable to find Mitre Enrichment for MitreID {mitre_id}")
 
-
-    def addMitreID(self, technique:dict, tactics:list[str], groups:list[str])->None:
-        
+    def addMitreIDViaGroupNames(self, technique:dict, tactics:list[str], groupNames:list[str])->None:
         technique_id = technique['technique_id']
         technique_obj = technique['technique']
         tactics.sort()
-        groups.sort()
-
+        
         if technique_id in self.data:
             raise Exception(f"Error, trying to redefine MITRE ID '{technique_id}'")
+        self.data[technique_id] = MitreAttackEnrichment(mitre_attack_id=technique_id, 
+                                                        mitre_attack_technique=technique_obj, 
+                                                        mitre_attack_tactics=tactics, 
+                                                        mitre_attack_groups=groupNames,
+                                                        mitre_attack_group_objects=[]) 
+
+    def addMitreIDViaGroupObjects(self, technique:dict, tactics:list[str],  groupObjects:list[dict[str,Any]])->None:
+        technique_id = technique['technique_id']
+        technique_obj = technique['technique']
+        tactics.sort()
 
+        groupNames:list[str] = sorted([group['group'] for group in groupObjects])
+        
+        if technique_id in self.data:
+            raise Exception(f"Error, trying to redefine MITRE ID '{technique_id}'")
         self.data[technique_id] = MitreAttackEnrichment(mitre_attack_id=technique_id, 
                                                         mitre_attack_technique=technique_obj, 
                                                         mitre_attack_tactics=tactics, 
-                                                        mitre_attack_groups=groups)
+                                                        mitre_attack_groups=groupNames,
+                                                        mitre_attack_group_objects=groupObjects)
 
 
     def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cached_or_offline: bool = False, skip_enrichment:bool = False) -> dict:
@@ -86,19 +98,20 @@ def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cach
                 progress_percent = ((index+1)/len(all_enterprise_techniques)) * 100
                 if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()):
                     print(f"\r\t{'MITRE Technique Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
-                apt_groups = []
+                apt_groups:list[dict[str,Any]] = []
                 for relationship in enterprise_relationships:
                     if (relationship['target_object'] == technique['id']) and relationship['source_object'].startswith('intrusion-set'):
                         for group in enterprise_groups:
                             if relationship['source_object'] == group['id']:
-                                apt_groups.append(group['group'])
+                                apt_groups.append(group)
+                                #apt_groups.append(group['group'])
 
                 tactics = []
                 if ('tactic' in technique):
                     for tactic in technique['tactic']:
                         tactics.append(tactic.replace('-',' ').title())
 
-                self.addMitreID(technique, tactics, apt_groups)
+                self.addMitreIDViaGroupObjects(technique, tactics, apt_groups)
                 attack_lookup[technique['technique_id']] = {'technique': technique['technique'], 'tactics': tactics, 'groups': apt_groups}
 
             if store_csv:
@@ -131,7 +144,7 @@ def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cach
                 technique_input = {'technique_id': key , 'technique': attack_lookup[key]['technique'] }
                 tactics_input = attack_lookup[key]['tactics']
                 groups_input = attack_lookup[key]['groups']
-                self.addMitreID(technique=technique_input, tactics=tactics_input, groups=groups_input)
+                self.addMitreIDViaGroupNames(technique=technique_input, tactics=tactics_input, groups=groups_input)