Merge branch 'main' into snapattack_datasource_enrichments

Author: ljstella

contentctl/objects/annotated_types.py

@@ -1,6 +1,9 @@
-from pydantic import Field
 from typing import Annotated
 
+from pydantic import Field
+
 CVE_TYPE = Annotated[str, Field(pattern=r"^CVE-[1|2]\d{3}-\d+$")]
-MITRE_ATTACK_ID_TYPE = Annotated[str, Field(pattern=r"^T\d{4}(.\d{3})?$")]
+MITRE_ATTACK_ID_TYPE_PARENT = Annotated[str, Field(pattern=r"^T\d{4}$")]
+MITRE_ATTACK_ID_TYPE_SUBTYPE = Annotated[str, Field(pattern=r"^T\d{4}(.\d{3})$")]
+MITRE_ATTACK_ID_TYPE = MITRE_ATTACK_ID_TYPE_PARENT | MITRE_ATTACK_ID_TYPE_SUBTYPE
 APPID_TYPE = Annotated[str, Field(pattern="^[a-zA-Z0-9_-]+$")]

contentctl/objects/constants.py

@@ -123,6 +123,8 @@
 CONTENTCTL_DETECTION_STANZA_NAME_FORMAT_TEMPLATE = (
     "{app_label} - {detection_name} - Rule"
 )
+
+CONTENTCTL_DASHBOARD_LABEL_TEMPLATE = "{app_label} - {dashboard_name}"
 CONTENTCTL_BASELINE_STANZA_NAME_FORMAT_TEMPLATE = "{app_label} - {detection_name}"
 CONTENTCTL_RESPONSE_TASK_NAME_FORMAT_TEMPLATE = (
     "{app_label} - {detection_name} - Response Task"

contentctl/objects/dashboard.py

@@ -1,14 +1,16 @@
+import json
+import pathlib
+from enum import StrEnum
 from typing import Any
-from pydantic import Field, Json, model_validator
 
-import pathlib
 from jinja2 import Environment
-import json
-from contentctl.objects.security_content_object import SecurityContentObject
+from pydantic import Field, Json, model_validator
+
 from contentctl.objects.config import build
-from enum import StrEnum
+from contentctl.objects.constants import CONTENTCTL_DASHBOARD_LABEL_TEMPLATE
+from contentctl.objects.security_content_object import SecurityContentObject
 
-DEFAULT_DASHBAORD_JINJA2_TEMPLATE = """<dashboard version="2" theme="{{ dashboard.theme }}">
+DEFAULT_DASHBOARD_JINJA2_TEMPLATE = """<dashboard version="2" theme="{{ dashboard.theme }}">
     <label>{{ dashboard.label(config) }}</label>
     <description></description>
     <definition><![CDATA[
@@ -31,7 +33,7 @@ class DashboardTheme(StrEnum):
 
 class Dashboard(SecurityContentObject):
     j2_template: str = Field(
-        default=DEFAULT_DASHBAORD_JINJA2_TEMPLATE,
+        default=DEFAULT_DASHBOARD_JINJA2_TEMPLATE,
         description="Jinja2 Template used to construct the dashboard",
     )
     description: str = Field(
@@ -49,7 +51,9 @@ class Dashboard(SecurityContentObject):
     )
 
     def label(self, config: build) -> str:
-        return f"{config.app.label} - {self.name}"
+        return CONTENTCTL_DASHBOARD_LABEL_TEMPLATE.format(
+            app_label=config.app.label, dashboard_name=self.name
+        )
 
     @model_validator(mode="before")
     @classmethod
@@ -98,7 +102,9 @@ 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()
+        # for clarity, the name of the dashboard file will follow the same convention
+        # as we use for detections, prefixing it with app_name -
+        filename = f"{self.label(config)}.xml"
         return pathlib.Path("default/data/ui/views") / filename
 
     def writeDashboardFile(self, j2_env: Environment, config: build):

contentctl/objects/detection_tags.py

@@ -33,7 +33,10 @@
     SecurityContentProductName,
     SecurityDomain,
 )
-from contentctl.objects.mitre_attack_enrichment import MitreAttackEnrichment
+from contentctl.objects.mitre_attack_enrichment import (
+    MitreAttackEnrichment,
+    MitreAttackGroup,
+)
 
 
 class DetectionTags(BaseModel):
@@ -44,7 +47,7 @@ class DetectionTags(BaseModel):
     asset_type: AssetType = Field(...)
     group: list[str] = []
 
-    mitre_attack_id: List[MITRE_ATTACK_ID_TYPE] = []
+    mitre_attack_id: list[MITRE_ATTACK_ID_TYPE] = []
     nist: list[NistCategory] = []
 
     product: list[SecurityContentProductName] = Field(..., min_length=1)
@@ -68,6 +71,15 @@ def kill_chain_phases(self) -> list[KillChainPhase]:
                 phases.add(phase)
         return sorted(list(phases))
 
+    # We do not want this to be included in serialization. By default, @property
+    # objects are not included in dumps
+    @property
+    def unique_mitre_attack_groups(self) -> list[MitreAttackGroup]:
+        group_set: set[MitreAttackGroup] = set()
+        for enrichment in self.mitre_attack_enrichments:
+            group_set.update(set(enrichment.mitre_attack_group_objects))
+        return sorted(group_set, key=lambda k: k.group)
+
     # enum is intentionally Cis18 even though field is named cis20 for legacy reasons
     @computed_field
     @property
@@ -134,8 +146,8 @@ def addAttackEnrichment(self, info: ValidationInfo):
 
         if len(missing_tactics) > 0:
             raise ValueError(f"Missing Mitre Attack IDs. {missing_tactics} not found.")
-        else:
-            self.mitre_attack_enrichments = mitre_enrichments
+
+        self.mitre_attack_enrichments = mitre_enrichments
 
         return self
 
@@ -159,6 +171,44 @@ def addAttackEnrichments(cls, v:list[MitreAttackEnrichment], info:ValidationInfo
         return enrichments
     """
 
+    @field_validator("mitre_attack_id", mode="after")
+    @classmethod
+    def sameTypeAndSubtypeNotPresent(
+        cls, techniques_and_subtechniques: list[MITRE_ATTACK_ID_TYPE]
+    ) -> list[MITRE_ATTACK_ID_TYPE]:
+        techniques: list[str] = [
+            f"{unknown_technique}."
+            for unknown_technique in techniques_and_subtechniques
+            if "." not in unknown_technique
+        ]
+        subtechniques: list[MITRE_ATTACK_ID_TYPE] = [
+            unknown_technique
+            for unknown_technique in techniques_and_subtechniques
+            if "." in unknown_technique
+        ]
+        subtype_and_parent_exist_exceptions: list[ValueError] = []
+
+        for subtechnique in subtechniques:
+            for technique in techniques:
+                if subtechnique.startswith(technique):
+                    subtype_and_parent_exist_exceptions.append(
+                        ValueError(
+                            f"    Technique   : {technique.split('.')[0]}\n"
+                            f"    SubTechnique: {subtechnique}\n"
+                        )
+                    )
+
+        if len(subtype_and_parent_exist_exceptions):
+            error_string = "\n".join(
+                str(e) for e in subtype_and_parent_exist_exceptions
+            )
+            raise ValueError(
+                "Overlapping MITRE Attack ID Techniques and Subtechniques may not be defined. "
+                f"Remove the Technique and keep the Subtechnique:\n{error_string}"
+            )
+
+        return techniques_and_subtechniques
+
     @field_validator("analytic_story", mode="before")
     @classmethod
     def mapStoryNamesToStoryObjects(
@@ -238,3 +288,6 @@ def mapAtomicGuidsToAtomicTests(
         return matched_tests + [
             AtomicTest.AtomicTestWhenTestIsMissing(test) for test in missing_tests
         ]
+        return matched_tests + [
+            AtomicTest.AtomicTestWhenTestIsMissing(test) for test in missing_tests
+        ]

contentctl/objects/mitre_attack_enrichment.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
-from pydantic import BaseModel, Field, ConfigDict, HttpUrl, field_validator
-from typing import List
-from enum import StrEnum
+
 import datetime
+from enum import StrEnum
+from typing import List
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator
+
 from contentctl.objects.annotated_types import MITRE_ATTACK_ID_TYPE
 
 
@@ -84,6 +87,16 @@ def standardize_contributors(cls, contributors: list[str] | None) -> list[str]:
             return []
         return contributors
 
+    def __lt__(self, other: MitreAttackGroup) -> bool:
+        if not isinstance(object, MitreAttackGroup):
+            raise Exception(
+                f"Cannot compare object of type MitreAttackGroup to object of type [{type(object).__name__}]"
+            )
+        return self.group < other.group
+
+    def __hash__(self) -> int:
+        return hash(self.group)
+
 
 class MitreAttackEnrichment(BaseModel):
     ConfigDict(extra="forbid")

contentctl/objects/story_tags.py

@@ -1,17 +1,18 @@
 from __future__ import annotations
-from pydantic import BaseModel, Field, model_serializer, ConfigDict
-from typing import List, Set, Optional
 
 from enum import Enum
+from typing import List, Optional, Set
 
-from contentctl.objects.mitre_attack_enrichment import MitreAttackEnrichment
+from pydantic import BaseModel, ConfigDict, Field, model_serializer
+
+from contentctl.objects.annotated_types import CVE_TYPE, MITRE_ATTACK_ID_TYPE
 from contentctl.objects.enums import (
-    StoryCategory,
     DataModel,
     KillChainPhase,
     SecurityContentProductName,
+    StoryCategory,
 )
-from contentctl.objects.annotated_types import CVE_TYPE, MITRE_ATTACK_ID_TYPE
+from contentctl.objects.mitre_attack_enrichment import MitreAttackEnrichment
 
 
 class StoryUseCase(str, Enum):

contentctl/templates/app_template/default/data/ui/nav/default.xml

@@ -1,7 +1,7 @@
 <nav search_view="search" color="#65A637">
   <view name="escu_summary" default="true"/>
-  <view name="feedback"/>
   <view name="search"/>
-  <view name="dashboards"/>
-  <a href="http://docs.splunk.com/Documentation/ESSOC">Docs</a>
+  <collection label="Dashboards">
+    <view source="unclassified" match=" - "/>
+  </collection>
 </nav>

contentctl/templates/detections/endpoint/anomalous_usage_of_7zip.yml

@@ -60,7 +60,6 @@ tags:
   asset_type: Endpoint
   mitre_attack_id:
   - T1560.001
-  - T1560
   product:
   - Splunk Enterprise
   - Splunk Enterprise Security

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "contentctl"
-version = "5.0.2"
+version = "5.0.4"
 
 description = "Splunk Content Control Tool"
 authors = ["STRT <[email protected]>"]