DRAFT: new RBA Object - Step 3 - ESCU 5.0

.github/workflows/test_against_escu.yml

@@ -35,6 +35,7 @@ jobs:
         with: 
           path: security_content
           repository: splunk/security_content
+          ref: rba_migration
 
       #Install the given version of Python we will test against
       - name: Install Required Python Version

contentctl/actions/build.py

@@ -10,11 +10,11 @@
 from contentctl.output.conf_writer import ConfWriter
 from contentctl.output.api_json_output import ApiJsonOutput
 from contentctl.output.data_source_writer import DataSourceWriter
-from contentctl.objects.lookup import Lookup
+from contentctl.objects.lookup import CSVLookup, Lookup_Type
 import pathlib
 import json
 import datetime
-from typing import Union
+import uuid
 
 from contentctl.objects.config import build
 
@@ -34,27 +34,41 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:
             updated_conf_files:set[pathlib.Path] = set()
             conf_output = ConfOutput(input_dto.config)
 
+
+            # Construct a path to a YML that does not actually exist.
+            # We mock this "fake" path since the YML does not exist.
+            # This ensures the checking for the existence of the CSV is correct
+            data_sources_fake_yml_path = input_dto.config.getPackageDirectoryPath() / "lookups" / "data_sources.yml"
+
             # Construct a special lookup whose CSV is created at runtime and
-            # written directly into the output folder. It is created with model_construct,
-            # not model_validate, because the CSV does not exist yet.
+            # written directly into the lookups folder. We will delete this after a build, 
+            # assuming that it is successful.
             data_sources_lookup_csv_path = input_dto.config.getPackageDirectoryPath() / "lookups" / "data_sources.csv"
-            DataSourceWriter.writeDataSourceCsv(input_dto.director_output_dto.data_sources, data_sources_lookup_csv_path)
-            input_dto.director_output_dto.addContentToDictMappings(Lookup.model_construct(description= "A lookup file that will contain the data source objects for detections.", 
-                                                                        filename=data_sources_lookup_csv_path, 
-                                                                        name="data_sources"))
 
+
+
+            DataSourceWriter.writeDataSourceCsv(input_dto.director_output_dto.data_sources, data_sources_lookup_csv_path)
+            input_dto.director_output_dto.addContentToDictMappings(CSVLookup.model_construct(name="data_sources",
+                                                                            id=uuid.UUID("b45c1403-6e09-47b0-824f-cf6e44f15ac8"),
+                                                                            version=1,
+                                                                            author=input_dto.config.app.author_name,
+                                                                            date = datetime.date.today(), 
+                                                                            description= "A lookup file that will contain the data source objects for detections.",
+                                                                            lookup_type=Lookup_Type.csv,
+                                                                            file_path=data_sources_fake_yml_path))
             updated_conf_files.update(conf_output.writeHeaders())
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.detections, SecurityContentType.detections))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.stories, SecurityContentType.stories))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.baselines, SecurityContentType.baselines))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.investigations, SecurityContentType.investigations))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.lookups, SecurityContentType.lookups))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.macros, SecurityContentType.macros))
-            updated_conf_files.update(conf_output.writeObjects(input_dto.director_output_dto.dashboards, SecurityContentType.dashboards))
+            updated_conf_files.update(conf_output.writeLookups(input_dto.director_output_dto.lookups))
+            updated_conf_files.update(conf_output.writeDetections(input_dto.director_output_dto.detections))
+            updated_conf_files.update(conf_output.writeStories(input_dto.director_output_dto.stories))
+            updated_conf_files.update(conf_output.writeBaselines(input_dto.director_output_dto.baselines))
+            updated_conf_files.update(conf_output.writeInvestigations(input_dto.director_output_dto.investigations))
+            updated_conf_files.update(conf_output.writeMacros(input_dto.director_output_dto.macros))
+            updated_conf_files.update(conf_output.writeDashboards(input_dto.director_output_dto.dashboards))
             updated_conf_files.update(conf_output.writeMiscellaneousAppFiles())
 
 
 
+
             #Ensure that the conf file we just generated/update is syntactically valid
             for conf_file in updated_conf_files:
                 ConfWriter.validateConfFile(conf_file) 
@@ -67,17 +81,15 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:
         if input_dto.config.build_api:    
             shutil.rmtree(input_dto.config.getAPIPath(), ignore_errors=True)
             input_dto.config.getAPIPath().mkdir(parents=True)
-            api_json_output = ApiJsonOutput()
-            for output_objects, output_type in [(input_dto.director_output_dto.detections, SecurityContentType.detections),
-                                                (input_dto.director_output_dto.stories, SecurityContentType.stories),
-                                                (input_dto.director_output_dto.baselines, SecurityContentType.baselines),
-                                                (input_dto.director_output_dto.investigations, SecurityContentType.investigations),
-                                                (input_dto.director_output_dto.lookups, SecurityContentType.lookups),
-                                                (input_dto.director_output_dto.macros, SecurityContentType.macros),
-                                                (input_dto.director_output_dto.deployments, SecurityContentType.deployments)]:
-                api_json_output.writeObjects(output_objects, input_dto.config.getAPIPath(), input_dto.config.app.label, output_type )
-
-
+            api_json_output = ApiJsonOutput(input_dto.config.getAPIPath(), input_dto.config.app.label)
+            api_json_output.writeDetections(input_dto.director_output_dto.detections)
+            api_json_output.writeStories(input_dto.director_output_dto.stories)
+            api_json_output.writeBaselines(input_dto.director_output_dto.baselines)
+            api_json_output.writeInvestigations(input_dto.director_output_dto.investigations)
+            api_json_output.writeLookups(input_dto.director_output_dto.lookups)
+            api_json_output.writeMacros(input_dto.director_output_dto.macros)
+            api_json_output.writeDeployments(input_dto.director_output_dto.deployments)
+
 
             #create version file for sse api
             version_file = input_dto.config.getAPIPath()/"version.json"

contentctl/actions/detection_testing/infrastructures/DetectionTestingInfrastructure.py

@@ -1094,6 +1094,7 @@ def retry_search_until_timeout(
             job = self.get_conn().search(query=search, **kwargs)
             results = JSONResultsReader(job.results(output_mode="json"))
 
+            # TODO (cmcginley): @ljstella you're removing this ultimately, right?
             # Consolidate a set of the distinct observable field names
             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"
@@ -1121,7 +1122,10 @@ def retry_search_until_timeout(
                     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_risk_objects} are missing in the detection results")
+                        e = Exception(
+                            f"The risk object field(s) {missing_risk_objects} are missing in the "
+                            "detection results"
+                        )
                         test.result.set_job_content(
                             job.content,
                             self.infrastructure,
@@ -1137,6 +1141,8 @@ def retry_search_until_timeout(
                     # on a field.  In this case, the field will appear but will not contain any values
                     current_empty_fields: set[str] = set()
 
+                    # TODO (cmcginley): @ljstella is this something we're keeping for testing as
+                    #   well?
                     for field in observable_fields_set:
                         if result.get(field, 'null') == 'null':
                             if field in risk_object_fields_set:

contentctl/actions/validate.py

@@ -6,6 +6,7 @@
 from contentctl.enrichments.attack_enrichment import AttackEnrichment
 from contentctl.enrichments.cve_enrichment import CveEnrichment
 from contentctl.objects.atomic import AtomicEnrichment
+from contentctl.objects.lookup import FileBackedLookup
 from contentctl.helper.utils import Utils
 from contentctl.objects.data_source import DataSource
 from contentctl.helper.splunk_app import SplunkApp
@@ -64,7 +65,7 @@ def ensure_no_orphaned_files_in_lookups(self, repo_path:pathlib.Path, director_o
         lookupsDirectory = repo_path/"lookups"
 
         # Get all of the files referneced by Lookups
-        usedLookupFiles:list[pathlib.Path] = [lookup.filename for lookup in director_output_dto.lookups if lookup.filename is not None] + [lookup.file_path for lookup in director_output_dto.lookups if lookup.file_path is not None]
+        usedLookupFiles:list[pathlib.Path] = [lookup.filename for lookup in director_output_dto.lookups if isinstance(lookup, FileBackedLookup)] + [lookup.file_path for lookup in director_output_dto.lookups if lookup.file_path is not None]
 
         # Get all of the mlmodel and csv files in the lookups directory
         csvAndMlmodelFiles  = Utils.get_security_content_files_from_directory(lookupsDirectory, allowedFileExtensions=[".yml",".csv",".mlmodel"], fileExtensionsToReturn=[".csv",".mlmodel"])

contentctl/input/director.py

@@ -14,7 +14,7 @@
 from contentctl.objects.playbook import Playbook
 from contentctl.objects.deployment import Deployment
 from contentctl.objects.macro import Macro
-from contentctl.objects.lookup import Lookup
+from contentctl.objects.lookup import LookupAdapter, Lookup
 from contentctl.objects.atomic import AtomicEnrichment
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.data_source import DataSource
@@ -58,13 +58,12 @@ def addContentToDictMappings(self, content: SecurityContentObject):
                 f" - {content.file_path}\n"
                 f" - {self.name_to_content_map[content_name].file_path}"
             )
-        
+
         if content.id in self.uuid_to_content_map:
             raise ValueError(
                 f"Duplicate id '{content.id}' with paths:\n"
                 f" - {content.file_path}\n"
-                f" - {self.uuid_to_content_map[content.id].file_path}"
-            )
+                f" - {self.uuid_to_content_map[content.id].file_path}")
 
         if isinstance(content, Lookup):
             self.lookups.append(content)
@@ -157,7 +156,8 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
                 modelDict = YmlReader.load_file(file)
 
                 if contentType == SecurityContentType.lookups:
-                    lookup = Lookup.model_validate(modelDict, context={"output_dto":self.output_dto, "config":self.input_dto})
+                    lookup = LookupAdapter.validate_python(modelDict, context={"output_dto":self.output_dto, "config":self.input_dto})
+                    #lookup = Lookup.model_validate(modelDict, context={"output_dto":self.output_dto, "config":self.input_dto})
                     self.output_dto.addContentToDictMappings(lookup)
 
                 elif contentType == SecurityContentType.macros:

contentctl/input/new_content_questions.py

@@ -48,7 +48,7 @@ def get_questions_detection(cls) -> list[dict[str,Any]]:
             {
                 'type': 'checkbox',
                 'message': 'Your data source',
-                'name': 'data_source',
+                'name': 'data_sources',
                 #In the future, we should dynamically populate this from the DataSource Objects we have parsed from the data_sources directory
                 'choices': sorted(DataSource._value2member_map_ )