fix: add support for detection strategy>technique detection relationships in changelog_helper

Author: elucchesileon

mitreattack/diffStix/changelog_helper.py

@@ -108,7 +108,7 @@ def __init__(
         self.new = new
         self.show_key = show_key
         self.site_prefix = site_prefix
-        self.types = ["techniques", "software", "groups", "campaigns", "assets", "mitigations", "datasources", "datacomponents"]
+        self.types = ["techniques", "software", "groups", "campaigns", "assets", "mitigations", "datasources", "datacomponents", "detectionstrategies", "analytics", "logsources"]
         self.use_mitre_cti = use_mitre_cti
         self.verbose = verbose
         self.include_contributors = include_contributors
@@ -127,6 +127,9 @@ def __init__(
             "mitigations": "Mitigations",
             "datasources": "Data Sources",
             "datacomponents": "Data Components",
+            "detectionstrategies": "Detection Strategies",
+            "analytics": "Analytics",
+            "logsources": "Log Sources",
         }
 
         self.section_descriptions = {
@@ -462,45 +465,74 @@ def find_technique_detection_changes(self, new_stix_obj: dict, domain: str):
         stix_id = new_stix_obj["id"]
         all_old_domain_datasources = self.data["old"][domain]["attack_objects"]["datasources"]
         all_old_domain_datacomponents = self.data["old"][domain]["attack_objects"]["datacomponents"]
+        all_old_domain_detectionstrategies = self.data["old"][domain]["attack_objects"]["detectionstrategies"]
         all_new_domain_datasources = self.data["new"][domain]["attack_objects"]["datasources"]
         all_new_domain_datacomponents = self.data["new"][domain]["attack_objects"]["datacomponents"]
-        old_detections = {}
-        new_detections = {}
+        all_new_domain_detectionstrategies = self.data["new"][domain]["attack_objects"]["detectionstrategies"]
+
+        old_datacomponent_detections = {}
+        old_detectionstrategy_detections = {}
+        new_datacomponent_detections = {}
+        new_detectionstrategy_detections = {}
 
         for _, detection_relationship in self.data["old"][domain]["relationships"]["detections"].items():
             if detection_relationship.get("x_mitre_deprecated") or detection_relationship.get("revoked"):
                 continue
             if stix_id == detection_relationship["target_ref"]:
-                old_datacomponent_id = detection_relationship["source_ref"]
-                old_datacomponent = all_old_domain_datacomponents[old_datacomponent_id]
-                old_datasource_id = old_datacomponent["x_mitre_data_source_ref"]
-                old_datasource = all_old_domain_datasources[old_datasource_id]
-                old_datasource_attack_id = get_attack_id(stix_obj=old_datasource)
-                old_detections[
-                    old_datacomponent_id
-                ] = f"{old_datasource_attack_id}: {old_datasource['name']} ({old_datacomponent['name']})"
+                old_sourceref_id = detection_relationship["source_ref"]
+                if old_sourceref_id in all_old_domain_datacomponents:
+                    old_datacomponent = all_old_domain_datacomponents[old_sourceref_id]
+                    old_datasource_id = old_datacomponent["x_mitre_data_source_ref"]
+                    old_datasource = all_old_domain_datasources[old_datasource_id]
+                    old_datasource_attack_id = get_attack_id(stix_obj=old_datasource)
+                    old_datacomponent_detections[
+                        old_sourceref_id
+                    ] = f"{old_datasource_attack_id}: {old_datasource['name']} ({old_datacomponent['name']})"
+                if old_sourceref_id in all_old_domain_detectionstrategies:
+                    old_detectionstrategy = all_old_domain_detectionstrategies[old_sourceref_id]
+                    old_detectionstrategy_attack_id = get_attack_id(stix_obj=old_detectionstrategy)
+                    old_detectionstrategy_detections[
+                        old_sourceref_id
+                    ] = f"{old_detectionstrategy_attack_id}: {old_detectionstrategy['name']}"
 
         for _, detection_relationship in self.data["new"][domain]["relationships"]["detections"].items():
             if detection_relationship.get("x_mitre_deprecated") or detection_relationship.get("revoked"):
                 continue
             if stix_id == detection_relationship["target_ref"]:
-                new_datacomponent_id = detection_relationship["source_ref"]
-                new_datacomponent = all_new_domain_datacomponents[new_datacomponent_id]
-                new_datasource_id = new_datacomponent["x_mitre_data_source_ref"]
-                new_datasource = all_new_domain_datasources[new_datasource_id]
-                new_datasource_attack_id = get_attack_id(stix_obj=new_datasource)
-                new_detections[
-                    new_datacomponent_id
-                ] = f"{new_datasource_attack_id}: {new_datasource['name']} ({new_datacomponent['name']})"
-
-        shared_detections = old_detections.keys() & new_detections.keys()
-        brand_new_detections = new_detections.keys() - old_detections.keys()
-        dropped_detections = old_detections.keys() - new_detections.keys()
-
-        new_stix_obj["changelog_detections"] = {
-            "shared": sorted([f"{new_detections[stix_id]}" for stix_id in shared_detections]),
-            "new": sorted([f"{new_detections[stix_id]}" for stix_id in brand_new_detections]),
-            "dropped": sorted([f"{old_detections[stix_id]}" for stix_id in dropped_detections]),
+                new_sourceref_id = detection_relationship["source_ref"]
+                if new_sourceref_id in all_new_domain_datacomponents:
+                    new_datacomponent = all_new_domain_datacomponents[new_sourceref_id]
+                    new_datasource_id = new_datacomponent["x_mitre_data_source_ref"]
+                    new_datasource = all_new_domain_datasources[new_datasource_id]
+                    new_datasource_attack_id = get_attack_id(stix_obj=new_datasource)
+                    new_datacomponent_detections[
+                        new_sourceref_id
+                    ] = f"{new_datasource_attack_id}: {new_datasource['name']} ({new_datacomponent['name']})"
+                if new_sourceref_id in all_new_domain_detectionstrategies:
+                    new_detectionstrategy = all_new_domain_detectionstrategies[new_sourceref_id]
+                    new_detectionstrategy_attack_id = get_attack_id(stix_obj=new_detectionstrategy)
+                    new_detectionstrategy_detections[
+                        new_sourceref_id
+                    ] = f"{new_detectionstrategy_attack_id}: {new_detectionstrategy['name']}"
+
+        shared_datacomponent_detections = old_datacomponent_detections.keys() & new_datacomponent_detections.keys()
+        brand_new_datacomponent_detections = new_datacomponent_detections.keys() - old_datacomponent_detections.keys()
+        dropped_datacomponent_detections = old_datacomponent_detections.keys() - new_datacomponent_detections.keys()
+
+        new_stix_obj["changelog_datacomponent_detections"] = {
+            "shared": sorted([f"{new_datacomponent_detections[stix_id]}" for stix_id in shared_datacomponent_detections]),
+            "new": sorted([f"{new_datacomponent_detections[stix_id]}" for stix_id in brand_new_datacomponent_detections]),
+            "dropped": sorted([f"{old_datacomponent_detections[stix_id]}" for stix_id in dropped_datacomponent_detections]),
+        }
+
+        shared_detectionstrategy_detections = old_detectionstrategy_detections.keys() & new_detectionstrategy_detections.keys()
+        brand_new_detectionstrategy_detections = new_detectionstrategy_detections.keys() - old_detectionstrategy_detections.keys()
+        dropped_detectionstrategy_detections = old_detectionstrategy_detections.keys() - new_detectionstrategy_detections.keys()
+
+        new_stix_obj["changelog_detectionstrategy_detections"] = {
+            "shared": sorted([f"{new_detectionstrategy_detections[stix_id]}" for stix_id in shared_detectionstrategy_detections]),
+            "new": sorted([f"{new_detectionstrategy_detections[stix_id]}" for stix_id in brand_new_detectionstrategy_detections]),
+            "dropped": sorted([f"{old_detectionstrategy_detections[stix_id]}" for stix_id in dropped_detectionstrategy_detections]),
         }
 
     def load_domain(self, domain: str):
@@ -589,6 +621,9 @@ def parse_extra_data(self, data_store: stix2.MemoryStore, domain: str, datastore
             "mitigations": [Filter("type", "=", "course-of-action")],
             "datasources": [Filter("type", "=", "x-mitre-data-source")],
             "datacomponents": [Filter("type", "=", "x-mitre-data-component")],
+            "detectionstrategies": [Filter("type", "=", "x-mitre-detection-strategy")],
+            "analytics": [Filter("type", "=", "x-mitre-analytic")],
+            "logsources": [Filter("type", "=", "x-mitre-log-source")],
         }
         for object_type, stix_filters in attack_type_to_stix_filter.items():
             raw_data = []
@@ -1669,17 +1704,32 @@ def write_detailed_html(html_file_detailed: str, diffStix: DiffStix):
                                     lines.append("</ul>")
 
                             # Detections!
-                            if stix_object.get("changelog_detections"):
-                                new_detections = stix_object["changelog_detections"].get("new")
-                                dropped_detections = stix_object["changelog_detections"].get("dropped")
+                            if stix_object.get("changelog_datacomponent_detections"):
+                                new_detections = stix_object["changelog_datacomponent_detections"].get("new")
+                                dropped_detections = stix_object["changelog_datacomponent_detections"].get("dropped")
+                                if new_detections:
+                                    lines.append("<p><b>New Detections (Data Components -> Technique)</b>:</p>")
+                                    lines.append("<ul>")
+                                    for detection in new_detections:
+                                        lines.append(f"  <li>{detection}</li>")
+                                    lines.append("</ul>")
+                                if dropped_detections:
+                                    lines.append("<p><b>Dropped Detections (Data Components -> Technique)</b>:</p>")
+                                    lines.append("<ul>")
+                                    for detection in dropped_detections:
+                                        lines.append(f"  <li>{detection}</li>")
+                                    lines.append("</ul>")
+                            if stix_object.get("changelog_detectionstrategy_detections"):
+                                new_detections = stix_object["changelog_detectionstrategy_detections"].get("new")
+                                dropped_detections = stix_object["changelog_detectionstrategy_detections"].get("dropped")
                                 if new_detections:
-                                    lines.append("<p><b>New Detections</b>:</p>")
+                                    lines.append("<p><b>New Detections (Detection Strategies -> Technique)</b>:</p>")
                                     lines.append("<ul>")
                                     for detection in new_detections:
                                         lines.append(f"  <li>{detection}</li>")
                                     lines.append("</ul>")
                                 if dropped_detections:
-                                    lines.append("<p><b>Dropped Detections</b>:</p>")
+                                    lines.append("<p><b>Dropped Detections (Detection Strategies -> Technique)</b>:</p>")
                                     lines.append("<ul>")
                                     for detection in dropped_detections:
                                         lines.append(f"  <li>{detection}</li>")