Updates in response to PR review

with colleague.

Author: pyth0n1c

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -40,6 +40,12 @@
 from contentctl.objects.enums import ProvidingTechnology
 from contentctl.enrichments.cve_enrichment import CveEnrichmentObj
 import datetime
+from contentctl.objects.constants import (
+    ES_MAX_STANZA_LENGTH,
+    ES_SEARCH_STANZA_NAME_FORMAT_AFTER_CLONING_IN_PRODUCT_TEMPLATE,
+    CONTENTCTL_MAX_SEARCH_NAME_LENGTH
+)
+
 MISSING_SOURCES: set[str] = set()
 
 # Those AnalyticsTypes that we do not test via contentctl
@@ -51,7 +57,7 @@
 # TODO (#266): disable the use_enum_values configuration
 class Detection_Abstract(SecurityContentObject):
     model_config = ConfigDict(use_enum_values=True)
-    name:str = Field(...,max_length=67)
+    name:str = Field(...,max_length=CONTENTCTL_MAX_SEARCH_NAME_LENGTH)
     #contentType: SecurityContentType = SecurityContentType.detections
     type: AnalyticsType = Field(...)
     status: DetectionStatus = Field(...)
@@ -74,26 +80,19 @@ class Detection_Abstract(SecurityContentObject):
 
     data_source_objects: list[DataSource] = []
 
-    def get_action_dot_correlationsearch_dot_label(self, app:CustomApp, max_stanza_length:int=99)->str:
-        label = self.get_conf_stanza_name(app)
-        label_after_saving_in_product = f"{self.tags.security_domain.value} - {label} - Rule"
-    
-        if len(label_after_saving_in_product) > max_stanza_length:
-            raise ValueError(f"label may only be {max_stanza_length} characters to allow updating in-product, "
-                             f"but stanza was actually {len(label_after_saving_in_product)} characters: '{label_after_saving_in_product}' ")
+    def get_action_dot_correlationsearch_dot_label(self, app:CustomApp, max_stanza_length:int=ES_MAX_STANZA_LENGTH)->str:
+        stanza_name = self.get_conf_stanza_name(app)
+        stanza_name_after_saving_in_es = ES_SEARCH_STANZA_NAME_FORMAT_AFTER_CLONING_IN_PRODUCT_TEMPLATE.format(
+            security_domain_value = self.tags.security_domain.value, 
+            search_name = stanza_name
+            )
 
-        return label
-    
-    def get_conf_stanza_name(self, app:CustomApp, max_stanza_length:int=81)->str:
-        stanza_name = f"{app.label} - {self.name} - Rule"
-        if len(stanza_name) > max_stanza_length:
-            raise ValueError(f"conf stanza may only be {max_stanza_length} characters, "
-                             f"but stanza was actually {len(stanza_name)} characters: '{stanza_name}' ")
-        #print(f"Stanza            Length[{len(stanza_name)}]")
-        return stanza_name
 
+        if len(stanza_name_after_saving_in_es) > max_stanza_length:
+            raise ValueError(f"label may only be {max_stanza_length} characters to allow updating in-product, "
+                             f"but stanza was actually {len(stanza_name_after_saving_in_es)} characters: '{stanza_name_after_saving_in_es}' ")
 
-    
+        return stanza_name    
 
     @field_validator("search", mode="before")
     @classmethod
@@ -674,7 +673,7 @@ def addTags_nist(self):
         else:
             self.tags.nist = [NistCategory.DE_AE]
         return self
-    
+        
 
     @model_validator(mode="after")
     def ensureThrottlingFieldsExist(self):
@@ -685,10 +684,6 @@ def ensureThrottlingFieldsExist(self):
         if self.tags.throttling is None:
             # No throttling configured for this detection
             return self
-        
-        if not isinstance(self.search, str):
-            # Search is sigma-formatted, so we cannot perform this validation.
-            return self
 
         missing_fields:list[str] = [field for field in self.tags.throttling.fields if field not in self.search]
         if len(missing_fields) > 0:

contentctl/objects/abstract_security_content_objects/security_content_object_abstract.py

@@ -5,8 +5,10 @@
     from contentctl.objects.deployment import Deployment
     from contentctl.objects.security_content_object import SecurityContentObject
     from contentctl.input.director import DirectorOutputDto
+    from contentctl.objects.config import CustomApp
 
 from contentctl.objects.enums import AnalyticsType
+from contentctl.objects.constants import CONTENTCTL_MAX_STANZA_LENGTH, CONTENTCTL_STANZA_NAME_FORMAT_TEMPLATE
 import abc
 import uuid
 import datetime
@@ -56,7 +58,14 @@ def serialize_model(self):
             "description": self.description,
             "references": [str(url) for url in self.references or []]
         }
-
+    
+    def get_conf_stanza_name(self, app:CustomApp, max_stanza_length:int=CONTENTCTL_MAX_STANZA_LENGTH)->str:
+        stanza_name = CONTENTCTL_STANZA_NAME_FORMAT_TEMPLATE.format(app_label=app.label, detection_name=self.name)
+        if len(stanza_name) > max_stanza_length:
+            raise ValueError(f"conf stanza may only be {max_stanza_length} characters, "
+                             f"but stanza was actually {len(stanza_name)} characters: '{stanza_name}' ")
+        return stanza_name
+    
     @staticmethod
     def objectListToNameList(objects: list[SecurityContentObject]) -> list[str]:
         return [object.getName() for object in objects]

contentctl/objects/baseline.py

@@ -6,16 +6,11 @@
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.enums import DataModel
 from contentctl.objects.baseline_tags import BaselineTags
-from contentctl.objects.config import CustomApp
-#from contentctl.objects.deployment import Deployment
-
-# from typing import TYPE_CHECKING
-# if TYPE_CHECKING:
-#     from contentctl.input.director import DirectorOutputDto
 
+from contentctl.objects.constants import CONTENTCTL_MAX_SEARCH_NAME_LENGTH
 
 class Baseline(SecurityContentObject):
-    name:str = Field(...,max_length=67)
+    name:str = Field(...,max_length=CONTENTCTL_MAX_SEARCH_NAME_LENGTH)
     type: Annotated[str,Field(pattern="^Baseline$")] = Field(...)
     datamodel: Optional[List[DataModel]] = None
     search: str = Field(..., min_length=4)
@@ -26,14 +21,6 @@ class Baseline(SecurityContentObject):
     # enrichment
     deployment: Deployment = Field({})
 
-    def get_conf_stanza_name(self, app:CustomApp, max_stanza_length:int=81)->str:
-        stanza_name = f"{app.label} - {self.name}"
-        if len(stanza_name) > max_stanza_length:
-            raise ValueError(f"conf stanza may only be {max_stanza_length} characters, "
-                             f"but stanza was actually {len(stanza_name)} characters: '{stanza_name}' ")
-        #print(f"Stanza            Length[{len(stanza_name)}]")
-        return stanza_name
-    
 
     @field_validator("deployment", mode="before")
     def getDeployment(cls, v:Any, info:ValidationInfo)->Deployment:

contentctl/objects/constants.py

@@ -1,3 +1,5 @@
+# Use for calculation of maximum length of name field
+from contentctl.objects.enums import SecurityDomain
 
 ATTACK_TACTICS_KILLCHAIN_MAPPING = {
     "Reconnaissance": "Reconnaissance",
@@ -140,3 +142,28 @@
 
 # The relative path to the directory where any apps/packages will be downloaded
 DOWNLOADS_DIRECTORY = "downloads"
+
+# Maximum length of the name field for a search.
+# This number is derived from a limitation that exists in 
+# ESCU where a search cannot be edited, due to validation
+# errors, if its name is longer than 99 characters.
+# When an saved search is cloned in Enterprise Security User Interface,
+# it is wrapped in the following: 
+# {Detection.tags.security_domain.value} - {SEARCH_STANZA_NAME} - Rule
+# Similarly, when we generate the search stanza name in contentctl, it
+# is app.label - detection.name - Rule
+# However, in product the search name is:
+# {CustomApp.label} - {detection.name} - Rule,
+# or in ESCU:
+# ESCU - {detection.name} - Rule,
+# this gives us a maximum length below.
+# When an ESCU search is cloned, it will 
+# have a full name like (the following is NOT a typo):
+# Endpoint - ESCU - Name of Search From YML File - Rule - Rule
+# The math below accounts for all these caveats
+ES_MAX_STANZA_LENGTH = 99
+CONTENTCTL_STANZA_NAME_FORMAT_TEMPLATE = "{app_label} - {detection_name} - Rule"
+ES_SEARCH_STANZA_NAME_FORMAT_AFTER_CLONING_IN_PRODUCT_TEMPLATE = "{security_domain_value} - {search_name} - Rule"
+SECURITY_DOMAIN_MAX_LENGTH = max([len(SecurityDomain[value]) for value in SecurityDomain._member_map_])
+CONTENTCTL_MAX_STANZA_LENGTH = ES_MAX_STANZA_LENGTH - len(ES_SEARCH_STANZA_NAME_FORMAT_AFTER_CLONING_IN_PRODUCT_TEMPLATE.format(security_domain_value="X"*SECURITY_DOMAIN_MAX_LENGTH,search_name=""))
+CONTENTCTL_MAX_SEARCH_NAME_LENGTH = CONTENTCTL_MAX_STANZA_LENGTH - len(CONTENTCTL_STANZA_NAME_FORMAT_TEMPLATE.format(app_label="ESCU", detection_name=""))

contentctl/objects/dashboard.py

@@ -2,13 +2,13 @@
 from pydantic import Field, Json, model_validator
 
 import pathlib
-import copy
 from jinja2 import Environment
 import json
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.config import build
+from enum import StrEnum
 
-DEFAULT_DASHBAORD_JINJA2_TEMPLATE = '''<dashboard version="2" theme="light">
+DEFAULT_DASHBAORD_JINJA2_TEMPLATE = '''<dashboard version="2" theme="{{ dashboard.theme }}">
     <label>{{ dashboard.label(config) }}</label>
     <description></description>
     <definition><![CDATA[
@@ -23,11 +23,15 @@
     ]]></meta>
 </dashboard>'''
 
+class DashboardTheme(StrEnum):
+    light = "light"
+    dark = "dark"
+
 class Dashboard(SecurityContentObject):
     j2_template: str = Field(default=DEFAULT_DASHBAORD_JINJA2_TEMPLATE, description="Jinja2 Template used to construct the dashboard")
     description: str = Field(...,description="A description of the dashboard. This does not have to match "
                              "the description of the dashboard in the JSON file.", max_length=10000)
-
+    theme: DashboardTheme = Field(default=DashboardTheme.dark, description="The theme of the dashboard. Choose between 'light' and 'dark'.")
     json_obj: Json[dict[str,Any]] = Field(..., description="Valid JSON object that describes the dashboard")
 
 

contentctl/objects/investigation.py

@@ -5,14 +5,15 @@
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.enums import DataModel
 from contentctl.objects.investigation_tags import InvestigationTags
-from contentctl.objects.config import CustomApp
+
+from contentctl.objects.constants import CONTENTCTL_MAX_SEARCH_NAME_LENGTH
 
 # TODO (#266): disable the use_enum_values configuration
 class Investigation(SecurityContentObject):
     model_config = ConfigDict(use_enum_values=True,validate_default=False)
     type: str = Field(...,pattern="^Investigation$")
     datamodel: list[DataModel] = Field(...)
-    name:str = Field(...,max_length=67)
+    name:str = Field(...,max_length=CONTENTCTL_MAX_SEARCH_NAME_LENGTH)
     search: str = Field(...)
     how_to_implement: str = Field(...)
     known_false_positives: str = Field(...)
@@ -69,13 +70,5 @@ def model_post_init(self, ctx:dict[str,Any]):
         for story in self.tags.analytic_story:
             story.investigations.append(self)
 
-    def get_conf_stanza_name(self, app:CustomApp, max_stanza_length:int=81)->str:
-        stanza_name = f"{app.label} - {self.name} - Response Task"
-        if len(stanza_name) > max_stanza_length:
-            raise ValueError(f"conf stanza may only be {max_stanza_length} characters, "
-                             f"but stanza was actually {len(stanza_name)} characters: '{stanza_name}' ")
-        #print(f"Stanza            Length[{len(stanza_name)}]")
-        return stanza_name
-    
 
 

contentctl/output/conf_writer.py

@@ -113,6 +113,11 @@ def writeDashboardFiles(config:build, dashboards:list[Dashboard])->set[pathlib.P
         written_files:set[pathlib.Path] = set()
         for dashboard in dashboards:
             output_file_path = dashboard.getOutputFilepathRelativeToAppRoot(config)
+            # Check that the full output path does not exist so that we are not having an
+            # name collision with a file in app_template
+            if (config.getPackageDirectoryPath()/output_file_path).exists():
+                raise FileExistsError(f"ERROR: Overwriting Dashboard File {output_file_path}. Does this file exist in {config.getAppTemplatePath()} AND {config.path/'dashboards'}?")
+                
             ConfWriter.writeXmlFileHeader(output_file_path, config)
             dashboard.writeDashboardFile(ConfWriter.getJ2Environment(), config)
             ConfWriter.validateXmlFile(config.getPackageDirectoryPath()/output_file_path)