Fix serialization issue with drilldowns.

Format of multiple drilldowns in
savedsearches.conf is now correct.
We are still populating the default
drilldowns, this feature will
eventually be removed.

Author: pyth0n1c

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -570,11 +570,21 @@ def model_post_init(self, __context: Any) -> None:
         # Update the search fields with the original search, if required
         for drilldown in self.drilldown_searches:
             drilldown.perform_search_substitutions(self)
-        print("adding default drilldown?")
+        
+        #For experimental purposes, add the default drilldowns
         self.drilldown_searches.extend(Drilldown.constructDrilldownsFromDetection(self))
 
     @property
-    def drilldownsInJSON(self) -> list[dict[str,str]]:
+    def drilldowns_in_JSON(self) -> list[dict[str,str]]:
+        """This function is required for proper JSON 
+        serializiation of drilldowns to occur in savedsearches.conf.
+        It returns the list[Drilldown] as a list[dict].
+        Without this function, the jinja template is unable
+        to convert list[Drilldown] to JSON
+
+        Returns:
+            list[dict[str,str]]: List of Drilldowns dumped to dict format
+        """        
         return [drilldown.model_dump() for drilldown in self.drilldown_searches]
 
     @field_validator('lookups', mode="before")

contentctl/objects/drilldown.py

@@ -1,23 +1,23 @@
 from __future__ import annotations
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_serializer
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from contentctl.objects.detection import Detection
 from contentctl.objects.enums import AnalyticsType
 SEARCH_PLACEHOLDER = "%original_detection_search%"
 EARLIEST_OFFSET = "$info_min_time$"
 LATEST_OFFSET = "$info_max_time$"
-RISK_SEARCH = "index = risk | stats count values(search_name) values(risk_message) values(analyticstories) values(annotations._all) values(annotations.mitre_attack.mitre_tactic) by risk_object"
+RISK_SEARCH = "index = risk  starthoursago = 168 endhoursago = 0 | stats count values(search_name) values(risk_message) values(analyticstories) values(annotations._all) values(annotations.mitre_attack.mitre_tactic) "
 
 class Drilldown(BaseModel):
     name: str = Field(..., description="The name of the drilldown search", min_length=5)
     search: str = Field(..., description="The text of a drilldown search. This must be valid SPL.", min_length=1)
-    earliest_offset:str = Field(..., 
+    earliest_offset:None | str = Field(..., 
                                 description="Earliest offset time for the drilldown search. "
                                 f"The most common value for this field is '{EARLIEST_OFFSET}', "
                                 "but it is NOT the default value and must be supplied explicitly.", 
                                 min_length= 1)
-    latest_offset:str = Field(..., 
+    latest_offset:None | str = Field(..., 
                               description="Latest offset time for the driolldown search. "
                               f"The most common value for this field is '{LATEST_OFFSET}', "
                               "but it is NOT the default value and must be supplied explicitly.", 
@@ -29,7 +29,7 @@ def constructDrilldownsFromDetection(cls, detection: Detection) -> list[Drilldow
         if len(victim_observables) == 0 or detection.type == AnalyticsType.Hunting:
             # No victims, so no drilldowns
             return []
-
+        print("Adding default drilldowns. REMOVE THIS BEFORE MERGING")
         variableNamesString = ' and '.join([f"${o.name}$" for o in victim_observables])
         nameField = f"View the detection results for {variableNamesString}"
         appendedSearch =  " | search " + ' '.join([f"{o.name} = ${o.name}$" for o in victim_observables])
@@ -38,8 +38,10 @@ def constructDrilldownsFromDetection(cls, detection: Detection) -> list[Drilldow
 
 
         nameField = f"View risk events for the last 7 days for {variableNamesString}"
-        search_field = f"{RISK_SEARCH}{appendedSearch}"
-        risk_events_last_7_days = cls(name=nameField, earliest_offset=EARLIEST_OFFSET, latest_offset=LATEST_OFFSET, search=search_field)
+        fieldNamesListString = ', '.join([o.name for o in victim_observables])
+        search_field = f"{RISK_SEARCH}by {fieldNamesListString} {appendedSearch}"
+        #risk_events_last_7_days = cls(name=nameField, earliest_offset=EARLIEST_OFFSET, latest_offset=LATEST_OFFSET, search=search_field)
+        risk_events_last_7_days = cls(name=nameField, earliest_offset=None, latest_offset=None, search=search_field)
 
         return [detection_results,risk_events_last_7_days]
 
@@ -50,4 +52,17 @@ def perform_search_substitutions(self, detection:Detection)->None:
                   f"drilldown search '{self.search}' for Detection {detection.file_path}.\n"
                   "If this was intentional, then please ignore this warning.\n")
         self.search = self.search.replace(SEARCH_PLACEHOLDER, detection.search)
-    
+    
+
+    @model_serializer
+    def serialize_model(self) -> dict[str,str]:
+        #Call serializer for parent
+        model:dict[str,str] = {}
+
+        model['name'] = self.name
+        model['search'] = self.search
+        if self.earliest_offset is not None:
+            model['earliest_offset'] = self.earliest_offset
+        if self.latest_offset is not None:
+            model['latest_offset'] = self.latest_offset
+        return model

contentctl/output/templates/savedsearches_detections.j2

@@ -112,7 +112,7 @@ alert.suppress.fields = {{ detection.tags.throttling.conf_formatted_fields() }}
 alert.suppress.period = {{ detection.tags.throttling.period }}
 {% endif %}
 search = {{ detection.search | escapeNewlines() }}
-action.notable.param.drilldown_searches = {{ detection.drilldownsInJSON | tojson | escapeNewlines() }}
+action.notable.param.drilldown_searches = {{ detection.drilldowns_in_JSON | tojson | escapeNewlines() }}
 {% endif %}
 
 {% endfor %}