Add suport for Data Source Objects and Event Source Objects

Author: Patrick

contentctl/actions/validate.py

@@ -6,33 +6,47 @@
 from typing import Union
 
 from contentctl.objects.enums import SecurityContentProduct
-from contentctl.objects.abstract_security_content_objects.security_content_object_abstract import SecurityContentObject_Abstract
-from contentctl.input.director import (
-    Director,
-    DirectorOutputDto
+from contentctl.objects.abstract_security_content_objects.security_content_object_abstract import (
+    SecurityContentObject_Abstract,
 )
+from contentctl.input.director import Director, DirectorOutputDto
 
 from contentctl.objects.config import validate
 from contentctl.enrichments.attack_enrichment import AttackEnrichment
 from contentctl.enrichments.cve_enrichment import CveEnrichment
 from contentctl.objects.atomic import AtomicTest
 
+
 class Validate:
     def execute(self, input_dto: validate) -> DirectorOutputDto:
-        
-        director_output_dto = DirectorOutputDto(AtomicTest.getAtomicTestsFromArtRepo(repo_path=input_dto.getAtomicRedTeamRepoPath(), 
-                                                                                     enabled=input_dto.enrichments),
-                                                AttackEnrichment.getAttackEnrichment(input_dto),
-                                                CveEnrichment.getCveEnrichment(input_dto),
-                                                [],[],[],[],[],[],[],[],[],[])
-
-        
+
+        director_output_dto = DirectorOutputDto(
+            AtomicTest.getAtomicTestsFromArtRepo(
+                repo_path=input_dto.getAtomicRedTeamRepoPath(),
+                enabled=input_dto.enrichments,
+            ),
+            AttackEnrichment.getAttackEnrichment(input_dto),
+            CveEnrichment.getCveEnrichment(input_dto),
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+            [],
+        )
+
         director = Director(director_output_dto)
         director.execute(input_dto)
-
         return director_output_dto
 
-    def validate_duplicate_uuids(self, security_content_objects:list[SecurityContentObject_Abstract]):
+    def validate_duplicate_uuids(
+        self, security_content_objects: list[SecurityContentObject_Abstract]
+    ):
         all_uuids = set()
         duplicate_uuids = set()
         for elem in security_content_objects:
@@ -45,14 +59,20 @@ def validate_duplicate_uuids(self, security_content_objects:list[SecurityContent
 
         if len(duplicate_uuids) == 0:
             return
-        
+
         # At least once duplicate uuid has been found. Enumerate all
         # the pieces of content that use duplicate uuids
         duplicate_messages = []
         for uuid in duplicate_uuids:
-            duplicate_uuid_content = [str(content.file_path) for content in security_content_objects if content.id in duplicate_uuids]
-            duplicate_messages.append(f"Duplicate UUID [{uuid}] in {duplicate_uuid_content}")
-        
+            duplicate_uuid_content = [
+                str(content.file_path)
+                for content in security_content_objects
+                if content.id in duplicate_uuids
+            ]
+            duplicate_messages.append(
+                f"Duplicate UUID [{uuid}] in {duplicate_uuid_content}"
+            )
+
         raise ValueError(
             "ERROR: Duplicate ID(s) found in objects:\n"
             + "\n - ".join(duplicate_messages)

contentctl/helper/utils.py

@@ -25,6 +25,9 @@ class Utils:
     @staticmethod
     def get_all_yml_files_from_directory(path: str) -> list[pathlib.Path]:
         listOfFiles:list[pathlib.Path] = []
+        base_path = pathlib.Path(path)
+        if not base_path.exists():
+            return listOfFiles
         for (dirpath, dirnames, filenames) in os.walk(path):
             for file in filenames:
                 if file.endswith(".yml"):
@@ -36,6 +39,8 @@ def get_all_yml_files_from_directory(path: str) -> list[pathlib.Path]:
     def get_all_yml_files_from_directory_one_layer_deep(path: str) -> list[pathlib.Path]:
         listOfFiles: list[pathlib.Path] = []
         base_path = pathlib.Path(path)
+        if not base_path.exists():
+            return listOfFiles
         # Check the base directory
         for item in base_path.iterdir():
             if item.is_file() and item.suffix == '.yml':

contentctl/input/director.py

@@ -22,6 +22,7 @@
 from contentctl.objects.atomic import AtomicTest
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.data_source import DataSource
+from contentctl.objects.event_source import EventSource
 
 from contentctl.enrichments.attack_enrichment import AttackEnrichment
 from contentctl.enrichments.cve_enrichment import CveEnrichment
@@ -57,6 +58,7 @@ class DirectorOutputDto:
     deployments: list[Deployment]
     ssa_detections: list[SSADetection]
     data_sources: list[DataSource]
+    event_sources: list[EventSource]
     name_to_content_map: dict[str, SecurityContentObject] = field(default_factory=dict)
     uuid_to_content_map: dict[UUID, SecurityContentObject] = field(default_factory=dict)
 
@@ -122,6 +124,7 @@ def execute(self, input_dto: validate) -> None:
         self.createSecurityContent(SecurityContentType.stories)
         self.createSecurityContent(SecurityContentType.baselines)
         self.createSecurityContent(SecurityContentType.investigations)
+        self.createSecurityContent(SecurityContentType.event_sources)
         self.createSecurityContent(SecurityContentType.data_sources)
         self.createSecurityContent(SecurityContentType.playbooks)
         self.createSecurityContent(SecurityContentType.detections)
@@ -141,6 +144,21 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
                 )
             )
 
+        elif contentType == SecurityContentType.event_sources:
+            security_content_files = Utils.get_all_yml_files_from_directory(
+                os.path.join(self.input_dto.path, "data_sources", "cloud", "event_sources")
+            )
+            security_content_files.extend(
+                Utils.get_all_yml_files_from_directory(
+                    os.path.join(self.input_dto.path, "data_sources", "endpoint", "event_sources")
+                )
+            )
+            security_content_files.extend(
+                Utils.get_all_yml_files_from_directory(
+                    os.path.join(self.input_dto.path, "data_sources", "network", "event_sources")
+                )
+            )
+
         elif contentType in [
             SecurityContentType.deployments,
             SecurityContentType.lookups,
@@ -183,12 +201,6 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
                         deployment = Deployment.model_validate(modelDict,context={"output_dto":self.output_dto})
                         self.output_dto.addContentToDictMappings(deployment)
 
-                elif contentType == SecurityContentType.data_sources:
-                    data_source = DataSource.model_validate(
-                        modelDict, context={"output_dto": self.output_dto}
-                    )
-                    self.output_dto.data_sources.append(data_source)
-
                 elif contentType == SecurityContentType.playbooks:
                         playbook = Playbook.model_validate(modelDict,context={"output_dto":self.output_dto})
                         self.output_dto.addContentToDictMappings(playbook)                  
@@ -214,6 +226,18 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
                         ssa_detection = self.ssa_detection_builder.getObject()
                         if ssa_detection.status in [DetectionStatus.production.value, DetectionStatus.validation.value]:
                             self.output_dto.addContentToDictMappings(ssa_detection)
+                
+                elif contentType == SecurityContentType.data_sources:
+                    data_source = DataSource.model_validate(
+                        modelDict, context={"output_dto": self.output_dto}
+                    )
+                    self.output_dto.data_sources.append(data_source)
+
+                elif contentType == SecurityContentType.event_sources:
+                    event_source = EventSource.model_validate(
+                        modelDict, context={"output_dto": self.output_dto}
+                    )
+                    self.output_dto.event_sources.append(event_source)
 
                 else:
                     raise Exception(f"Unsupported type: [{contentType}]")
@@ -262,7 +286,7 @@ def constructSSADetection(
         file_path: str,
     ) -> None:
         builder.reset()
-        builder.setObject(file_path, self.output_dto)
+        builder.setObject(file_path)
         builder.addMitreAttackEnrichmentNew(directorOutput.attack_enrichment)
         builder.addKillChainPhase()
         builder.addCIS()

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -369,8 +369,6 @@ def model_post_init(self, ctx:dict[str,Any]):
         # if not isinstance(director,DirectorOutputDto):
         #     raise ValueError("DirectorOutputDto was not passed in context of Detection model_post_init")
         director: Optional[DirectorOutputDto] = ctx.get("output_dto",None)
-        for story in self.tags.analytic_story:
-            story.detections.append(self)
 
         #Ensure that all baselines link to this detection
         for baseline in self.baselines:
@@ -399,6 +397,10 @@ def model_post_init(self, ctx:dict[str,Any]):
                 unique_data_sources[data_source_obj.name] = data_source_obj
         self.data_source_objects = list(unique_data_sources.values())
 
+        for story in self.tags.analytic_story:
+            story.detections.append(self)
+            story.data_sources.extend(self.data_source_objects)
+
         return self
 
 

contentctl/objects/data_source.py

@@ -12,5 +12,17 @@ class DataSource(BaseModel):
     configuration: str = None
     supported_TA: dict
     event_names: list = None
+    event_sources: list = None
     fields: list = None
-    example_log: str = None
+    example_log: str = None
+
+    def model_post_init(self, ctx:dict[str,Any]):
+        context = ctx.get("output_dto")
+        
+        if self.event_names:
+            self.event_sources = []
+            for event_source in context.event_sources:
+                if any(event['event_name'] == event_source.event_name for event in self.event_names):
+                    self.event_sources.append(event_source)
+
+        return self

contentctl/objects/enums.py

@@ -56,6 +56,7 @@ class SecurityContentType(enum.Enum):
     unit_tests = 9
     ssa_detections = 10
     data_sources = 11
+    event_sources = 12
 
 # Bringing these changes back in line will take some time after
 # the initial merge is complete

contentctl/objects/event_source.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+from pydantic import BaseModel
+
+
+class EventSource(BaseModel):
+    event_name: str
+    fields: list[str]
+    field_mappings: list[dict] = None
+    convert_to_log_source: list[dict] = None
+    example_log: str = None

contentctl/objects/story.py

@@ -7,7 +7,7 @@
     from contentctl.objects.detection import Detection
     from contentctl.objects.investigation import Investigation
     from contentctl.objects.baseline import Baseline
-    
+    from contentctl.objects.data_source import DataSource
 
 from contentctl.objects.security_content_object import SecurityContentObject
 
@@ -33,7 +33,7 @@ class Story(SecurityContentObject):
     detections:List[Detection] = []
     investigations: List[Investigation] = []
     baselines: List[Baseline] = []
-
+    data_sources: List[DataSource] = []
 
     def storyAndInvestigationNamesWithApp(self, app_name:str)->List[str]:
         return [f"{app_name} - {name} - Rule" for name in self.detection_names] + \