Dashboard Support

contentctl/actions/build.py

@@ -51,6 +51,7 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:
             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.writeAppConf())
 
             #Ensure that the conf file we just generated/update is syntactically valid

contentctl/actions/validate.py

@@ -10,7 +10,6 @@
 
 class Validate:
     def execute(self, input_dto: validate) -> DirectorOutputDto:
-
         director_output_dto = DirectorOutputDto(
             AtomicTest.getAtomicTestsFromArtRepo(
                 repo_path=input_dto.getAtomicRedTeamRepoPath(),
@@ -28,6 +27,7 @@ def execute(self, input_dto: validate) -> DirectorOutputDto:
             [],
             [],
             [],
+            []
         )
 
         director = Director(director_output_dto)

contentctl/input/director.py

@@ -20,6 +20,7 @@
 from contentctl.objects.lookup import Lookup
 from contentctl.objects.ssa_detection import SSADetection
 from contentctl.objects.atomic import AtomicTest
+from contentctl.objects.dashboard import Dashboard
 from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.data_source import DataSource
 from contentctl.objects.event_source import EventSource
@@ -56,6 +57,7 @@ class DirectorOutputDto:
     macros: list[Macro]
     lookups: list[Lookup]
     deployments: list[Deployment]
+    dashboards: list[Dashboard]
     ssa_detections: list[SSADetection]
     data_sources: list[DataSource]
     name_to_content_map: dict[str, SecurityContentObject] = field(default_factory=dict)
@@ -100,6 +102,8 @@ def addContentToDictMappings(self, content: SecurityContentObject):
             self.detections.append(content)
         elif isinstance(content, SSADetection):
             self.ssa_detections.append(content)
+        elif isinstance(content, Dashboard):
+            self.dashboards.append(content)            
         elif isinstance(content, DataSource):
             self.data_sources.append(content)
         else:
@@ -129,6 +133,7 @@ def execute(self, input_dto: validate) -> None:
         self.createSecurityContent(SecurityContentType.data_sources)
         self.createSecurityContent(SecurityContentType.playbooks)
         self.createSecurityContent(SecurityContentType.detections)
+        self.createSecurityContent(SecurityContentType.dashboards)
         self.createSecurityContent(SecurityContentType.ssa_detections)
 
 
@@ -143,6 +148,7 @@ def execute(self, input_dto: validate) -> None:
 
     def createSecurityContent(self, contentType: SecurityContentType) -> None:
         if contentType == SecurityContentType.ssa_detections:
+
             files = Utils.get_all_yml_files_from_directory(
                 os.path.join(self.input_dto.path, "ssa_detections")
             )
@@ -157,6 +163,7 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
             SecurityContentType.playbooks,
             SecurityContentType.detections,
             SecurityContentType.data_sources,
+            SecurityContentType.dashboards
         ]:
             files = Utils.get_all_yml_files_from_directory(
                 os.path.join(self.input_dto.path, str(contentType.name))
@@ -207,8 +214,13 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
                     self.output_dto.addContentToDictMappings(story)
 
                 elif contentType == SecurityContentType.detections:
+
                     detection = Detection.model_validate(modelDict,context={"output_dto":self.output_dto, "app":self.input_dto.app})
                     self.output_dto.addContentToDictMappings(detection)
+
+                elif contentType == SecurityContentType.dashboards:
+                        dashboard = Dashboard.model_validate(modelDict,context={"output_dto":self.output_dto})
+                        self.output_dto.addContentToDictMappings(dashboard)
 
                 elif contentType == SecurityContentType.ssa_detections:
                     self.constructSSADetection(self.ssa_detection_builder, self.output_dto,str(file))

contentctl/objects/dashboard.py

@@ -0,0 +1,96 @@
+from typing import Any
+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
+
+DEFAULT_DASHBAORD_JINJA2_TEMPLATE = '''<dashboard version="2" theme="light">
+    <label>{{ dashboard.label(config) }}</label>
+    <description></description>
+    <definition><![CDATA[
+{{ dashboard.pretty_print_json_obj() }}
+    ]]></definition>
+    <meta type="hiddenElements"><![CDATA[
+{
+	"hideEdit": false,
+	"hideOpenInSearch": false,
+	"hideExport": false
+}
+    ]]></meta>
+</dashboard>'''
+
+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)
+
+    json_obj: Json[dict[str,Any]] = Field(..., description="Valid JSON object that describes the dashboard")
+
+
+
+    def label(self, config:build)->str:
+        return f"{config.app.label} - {self.name}"
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_fields_from_json(cls, data:Any)->Any:
+        yml_file_name:str|None = data.get("file_path", None)
+        if yml_file_name is None:
+            raise ValueError("File name not passed to dashboard constructor")
+        yml_file_path = pathlib.Path(yml_file_name)
+        json_file_path = yml_file_path.with_suffix(".json")
+
+        if not json_file_path.is_file():
+            raise ValueError(f"Required file {json_file_path} does not exist.")
+
+        with open(json_file_path,'r') as jsonFilePointer:
+            try:
+                json_obj:dict[str,Any] = json.load(jsonFilePointer)
+            except Exception as e:
+                raise ValueError(f"Unable to load data from {json_file_path}: {str(e)}")
+
+        name_from_file = data.get("name",None)
+        name_from_json  = json_obj.get("title",None)
+
+        errors:list[str] = []
+        if name_from_json is None:
+            errors.append(f"'title' field is missing from {json_file_path}")
+        elif name_from_json != name_from_file:
+            errors.append(f"The 'title' field in the JSON file [{json_file_path}] does not match the 'name' field in the YML object [{yml_file_path}]. These two MUST match:\n    "
+                          f"title in JSON : {name_from_json}\n    "
+                          f"title in YML  : {name_from_file}\n    ")
+
+        description_from_json = json_obj.get("description",None)
+        if description_from_json is None:
+            errors.append("'description' field is missing from field 'json_object'")
+
+        if len(errors) > 0 :
+            err_string = "\n  - ".join(errors)
+            raise ValueError(f"Error(s) validating dashboard:\n  - {err_string}")
+
+        data['name'] = name_from_file
+        data['json_obj'] = json.dumps(json_obj)
+        return data
+
+
+    def pretty_print_json_obj(self):
+        return json.dumps(self.json_obj, indent=4)
+
+    def getOutputFilepathRelativeToAppRoot(self, config:build)->pathlib.Path:
+        filename = f"{self.file_path.stem}.xml".lower()
+        return pathlib.Path("default/data/ui/views")/filename
+
+
+    def writeDashboardFile(self, j2_env:Environment, config:build):
+        template = j2_env.from_string(self.j2_template)
+        dashboard_text = template.render(config=config, dashboard=self)
+
+        with open(config.getPackageDirectoryPath()/self.getOutputFilepathRelativeToAppRoot(config), 'a') as f:
+            output_xml = dashboard_text.encode('utf-8', 'ignore').decode('utf-8')
+            f.write(output_xml)
+
+

contentctl/objects/enums.py

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

contentctl/output/conf_output.py

@@ -152,6 +152,10 @@ def writeObjects(self, objects: list, type: SecurityContentType = None) -> set[p
                                     'macros.j2',
                                     self.config, objects))
 
+        elif type == SecurityContentType.dashboards:
+            written_files.update(ConfWriter.writeDashboardFiles(self.config, objects))
+
+
         return written_files
 
 

contentctl/output/conf_writer.py

@@ -8,6 +8,7 @@
 from jinja2 import Environment, FileSystemLoader, StrictUndefined
 import pathlib
 from contentctl.objects.security_content_object import SecurityContentObject
+from contentctl.objects.dashboard import Dashboard
 from contentctl.objects.config import build
 import xml.etree.ElementTree as ET
 
@@ -104,6 +105,17 @@ def writeXmlFile(app_output_path:pathlib.Path, template_name : str, config: buil
 
 
 
+    @staticmethod
+    def writeDashboardFiles(config:build, dashboards:list[Dashboard])->set[pathlib.Path]:
+        written_files:set[pathlib.Path] = set()
+        for dashboard in dashboards:
+            output_file_path = dashboard.getOutputFilepathRelativeToAppRoot(config)
+            ConfWriter.writeXmlFileHeader(output_file_path, config)
+            dashboard.writeDashboardFile(ConfWriter.getJ2Environment(), config)
+            ConfWriter.validateXmlFile(config.getPackageDirectoryPath()/output_file_path)
+            written_files.add(output_file_path)
+        return written_files
+
 
     @staticmethod
     def writeXmlFileHeader(app_output_path:pathlib.Path, config: build) -> None: