Able to build without any errors,

but that does not mean it is correct
yet. Outputs must be diffed against
prior versions to make sure there are
no unintended changes.

Author: pyth0n1c

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
-
+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.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.writeLookups(input_dto.director_output_dto.lookups))
             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) 

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

@@ -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)

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -16,7 +16,7 @@
 )
 
 from contentctl.objects.macro import Macro
-from contentctl.objects.lookup import Lookup
+from contentctl.objects.lookup import Lookup, FileBackedLookup, KVStoreLookup
 if TYPE_CHECKING:
     from contentctl.input.director import DirectorOutputDto
     from contentctl.objects.baseline import Baseline
@@ -285,10 +285,8 @@ def annotations(self) -> dict[str, Union[List[str], int, str]]:
 
         annotations_dict: dict[str, str | list[str] | int] = {}
         annotations_dict["analytic_story"] = [story.name for story in self.tags.analytic_story]
-        annotations_dict["confidence"] = self.tags.confidence
         if len(self.tags.cve or []) > 0:
             annotations_dict["cve"] = self.tags.cve
-        annotations_dict["impact"] = self.tags.impact
         annotations_dict["type"] = self.type
         annotations_dict["type_list"] = [self.type]
         # annotations_dict["version"] = self.version
@@ -480,6 +478,11 @@ def serialize_model(self):
             "source": self.source,
             "nes_fields": self.nes_fields,
         }
+        if self.rba is not None:
+            model["risk_severity"] = self.rba.severity
+            model['tags']['risk_score'] = self.rba.risk_score
+        else:
+            model['tags']['risk_score'] = 0
 
         # Only a subset of macro fields are required:
         all_macros: list[dict[str, str | list[str]]] = []
@@ -497,27 +500,26 @@ def serialize_model(self):
 
         all_lookups: list[dict[str, str | int | None]] = []
         for lookup in self.lookups:
-            if lookup.collection is not None:
+            if isinstance(lookup, KVStoreLookup):
                 all_lookups.append(
                     {
                         "name": lookup.name,
                         "description": lookup.description,
                         "collection": lookup.collection,
                         "case_sensitive_match": None,
-                        "fields_list": lookup.fields_list
+                        "fields_list": lookup.fields_to_fields_list_conf_format
                     }
                 )
-            elif lookup.filename is not None:
+            elif isinstance(lookup, FileBackedLookup):
                 all_lookups.append(
                     {
                         "name": lookup.name,
                         "description": lookup.description,
                         "filename": lookup.filename.name,
                         "default_match": "true" if lookup.default_match else "false",
                         "case_sensitive_match": "true" if lookup.case_sensitive_match else "false",
-                        "match_type": lookup.match_type,
-                        "min_matches": lookup.min_matches,
-                        "fields_list": lookup.fields_list
+                        "match_type": lookup.match_type_to_conf_format,
+                        "min_matches": lookup.min_matches
                     }
                 )
         model['lookups'] = all_lookups                                                              # type: ignore
@@ -790,7 +792,7 @@ def ensureProperRBAConfig(self):
         """
 
 
-        if self.deployment.alert_action.rba.enabled is False or self.deployment.alert_action.rba is None:
+        if self.deployment.alert_action.rba is None or self.deployment.alert_action.rba.enabled is False:
             # confirm we don't have an RBA config
             if self.rba is None:
                 return self

contentctl/objects/detection_tags.py

@@ -27,7 +27,6 @@
     Cis18Value,
     AssetType,
     SecurityDomain,
-    RiskSeverity,
     KillChainPhase,
     NistCategory,
     SecurityContentProductName
@@ -43,23 +42,6 @@ class DetectionTags(BaseModel):
     asset_type: AssetType = Field(...)
     group: list[str] = []
 
-    @computed_field
-    @property
-    def severity(self)->RiskSeverity:
-        if 0 <= self.risk_score <= 20:
-            return RiskSeverity.INFORMATIONAL
-        elif 20 < self.risk_score <= 40:
-            return RiskSeverity.LOW
-        elif 40 < self.risk_score <= 60:
-            return RiskSeverity.MEDIUM
-        elif 60 < self.risk_score <= 80:
-            return RiskSeverity.HIGH
-        elif 80 < self.risk_score <= 100:
-            return RiskSeverity.CRITICAL
-        else:
-            raise Exception(f"Error getting severity - risk_score must be between 0-100, but was actually {self.risk_score}")
-
-
     mitre_attack_id: List[MITRE_ATTACK_ID_TYPE] = []
     nist: list[NistCategory] = []
 
@@ -144,9 +126,7 @@ def serialize_model(self):
             "cis20": self.cis20,
             "kill_chain_phases": self.kill_chain_phases,
             "nist": self.nist,
-            "risk_score": self.risk_score,
             "security_domain": self.security_domain,
-            "risk_severity": self.severity,
             "mitre_attack_id": self.mitre_attack_id,
             "mitre_attack_enrichments": self.mitre_attack_enrichments
         }

contentctl/objects/lookup.py

@@ -80,7 +80,10 @@ def fix_lookup_path(cls, data:Any, info: ValidationInfo)->Any:
         return data
 
 
-    
+    @computed_field
+    @cached_property
+    def match_type_to_conf_format(self)->str:
+        return ', '.join(self.match_type)
 
 
     @staticmethod
@@ -196,6 +199,16 @@ def ensure_key(cls, values: list[str]):
             raise ValueError(f"fields MUST begin with '_key', not '{values[0]}'")
         return values
 
+    @computed_field
+    @cached_property
+    def collection(self)->str:
+        return self.name
+
+    @computed_field
+    @cached_property
+    def fields_to_fields_list_conf_format(self)->str:
+        return ', '.join(self.fields)
+
     @model_serializer
     def serialize_model(self):
         #Call parent serializer
@@ -204,7 +217,7 @@ def serialize_model(self):
         #All fields custom to this model
         model= {
             "collection": self.collection,
-            "fields_list": ", ".join(self.fields)
+            "fields_list": self.fields_to_fields_list_conf_format
         }
 
         #return the model

contentctl/objects/macro.py

@@ -48,7 +48,6 @@ def serialize_model(self):
         return model
 
     @staticmethod
-
     def get_macros(text_field:str, director:DirectorOutputDto , ignore_macros:set[str]=MACROS_TO_IGNORE)->list[Macro]:
         #Remove any comments, allowing there to be macros (which have a single backtick) inside those comments
         #If a comment ENDS in a macro, for example ```this is a comment with a macro `macro_here````
@@ -59,10 +58,10 @@ def get_macros(text_field:str, director:DirectorOutputDto , ignore_macros:set[st
                             "This may have occurred when a macro was commented out.\n"
                             "Please ammend your search to remove the substring '````'")
 
-        # replace all the macros with a space    
+        # Replace all the comments with a space. This prevents a comment from looking like a macro to the parser below    
         text_field = re.sub(r"\`\`\`[\s\S]*?\`\`\`", " ", text_field) 
 
-        
+        # Find all the macros, which start and end with a '`' character  
         macros_to_get = re.findall(r'`([^\s]+)`', text_field)
         #If macros take arguments, stop at the first argument.  We just want the name of the macro
         macros_to_get = set([macro[:macro.find('(')] if macro.find('(') != -1 else macro for macro in macros_to_get])

contentctl/objects/rba.py

@@ -1,9 +1,11 @@
 from enum import Enum
-from pydantic import BaseModel
+from pydantic import BaseModel, computed_field, Field
 from abc import ABC
-from typing import Set
+from typing import Set, Annotated
+from contentctl.objects.enums import RiskSeverity
 
 
+RiskScoreValue_Type = Annotated[int, Field(ge=1, le=100)]
 
 class RiskObjectType(str, Enum):
     SYSTEM = "system"
@@ -40,7 +42,7 @@ class ThreatObjectType(str, Enum):
 class risk_object(BaseModel):
     field: str
     type: RiskObjectType
-    score: int
+    score: RiskScoreValue_Type
 
     def __hash__(self):
         return hash((self.field, self.type, self.score))
@@ -54,5 +56,34 @@ def __hash__(self):
 
 class rba_object(BaseModel, ABC):
     message: str
-    risk_objects: Set[risk_object] 
-    threat_objects: Set[threat_object]
+    risk_objects: Annotated[Set[risk_object], Field(min_length=1)]
+    threat_objects: Set[threat_object]
+
+    
+    
+    @computed_field
+    @property
+    def risk_score(self)->RiskScoreValue_Type:
+        # First get the maximum score associated with
+        # a risk object. If there are no objects, then
+        # we should throw an exception.
+        if len(self.risk_objects) == 0:
+            raise Exception("There must be at least one Risk Object present to get Severity.")
+        return max([risk_object.score for risk_object in self.risk_objects])
+    
+    @computed_field
+    @property
+    def severity(self)->RiskSeverity:
+        if 0 <= self.risk_score <= 20:
+            return RiskSeverity.INFORMATIONAL
+        elif 20 < self.risk_score <= 40:
+            return RiskSeverity.LOW
+        elif 40 < self.risk_score <= 60:
+            return RiskSeverity.MEDIUM
+        elif 60 < self.risk_score <= 80:
+            return RiskSeverity.HIGH
+        elif 80 < self.risk_score <= 100:
+            return RiskSeverity.CRITICAL
+        else:
+            raise Exception(f"Error getting severity - risk_score must be between 0-100, but was actually {max_score}")
+

contentctl/output/api_json_output.py

@@ -215,7 +215,7 @@ def writeLookups(
                         "name",
                         "description",
                         "collection",
-                        "fields_list",
+                        "fields_to_fields_list_conf_format",
                         "filename",
                         "default_match",
                         "match_type",

contentctl/output/conf_output.py

@@ -1,34 +1,22 @@
 from __future__ import annotations
 from typing import TYPE_CHECKING, Callable
 if TYPE_CHECKING:
-    from contentctl.objects.security_content_object import SecurityContentObject
     from contentctl.objects.detection import Detection
-    from contentctl.objects.lookup import Lookup, FileBackedLookup
+    from contentctl.objects.lookup import Lookup
     from contentctl.objects.macro import Macro
     from contentctl.objects.dashboard import Dashboard
     from contentctl.objects.story import Story
     from contentctl.objects.baseline import Baseline
     from contentctl.objects.investigation import Investigation
 
-from dataclasses import dataclass
-import os
-import glob
+from contentctl.objects.lookup import FileBackedLookup
 import shutil
-import sys
 import tarfile
-from typing import Union
-from pathlib import Path
 import pathlib
-import time
 import timeit
 import datetime
-import shutil
-import json
 from contentctl.output.conf_writer import ConfWriter
-from contentctl.objects.enums import SecurityContentType
 from contentctl.objects.config import build
-from requests import Session, post, get
-from requests.auth import HTTPBasicAuth
 
 class ConfOutput:
     config: build