chore: update excel files

Author: adpare

mitreattack/attackToExcel/attackToExcel.py

@@ -148,24 +148,32 @@ def build_dataframes(src: MemoryStore, domain: str) -> Dict:
 
 
 def build_ds_an_lg_relationships(dataframes: Dict) -> Dict[str, pd.DataFrame]:
-    """Build sheets for ds-an-lg.xlsx using existing relationship tables."""
-    # Use existing DetectionStrategy -> Analytics relationship table
-    ds_an = dataframes["detectionstrategies"].get("detectionstrategies-analytic", pd.DataFrame())
+    """Build detection-mappings.xlsx with a single DS → Analytic → LogSource sheet."""
 
-    # Use existing Analytics -> LogSource relationship table
-    an_lg = dataframes["analytics"].get("analytic-logsource", pd.DataFrame())
+    ds_an = dataframes["detectionstrategies"].get(
+        "detectionstrategies-analytic", pd.DataFrame()
+    )
 
-    # Use existing Analytics -> Detection Strategy relationship table
-    an_ds = dataframes["analytics"].get("analytic-detectionstrategy", pd.DataFrame())
+    an_ls = dataframes["analytics"].get(
+        "analytic-logsource", pd.DataFrame()
+    )
+
+    if ds_an.empty or an_ls.empty:
+        combined = pd.DataFrame()
+    else:
+        combined = ds_an.merge(
+            an_ls,
+            on=["analytic_id", "analytic_name", "platforms"],
+            how="left",
+        )
 
     return {
-        "detectionstrategy_to_analytics": ds_an,
-        "analytics_to_logsources": an_lg,
-        "analytics_to_detectionstrategy": an_ds,
+        "ds_an_ls": combined
     }
 
 
-def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, output_dir: str = ".") -> List:
+
+def write_excel(dataframes: Dict, domain: str, src: MemoryStore, version: Optional[str] = None, output_dir: str = ".") -> List:
     """Given a set of dataframes from build_dataframes, write the ATT&CK dataset to output directory.
 
     Parameters
@@ -174,6 +182,9 @@ def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, ou
         A dictionary of pandas dataframes as built by build_dataframes()
     domain : str
         Domain of ATT&CK the dataframes correspond to, e.g "enterprise-attack"
+    src : stix2.MemoryStore
+        A STIX bundle containing ATT&CK data for a domain already loaded into memory.
+        Mutually exclusive with `remote` and `stix_file`.
     version : str, optional
         The version of ATT&CK the dataframes correspond to, e.g "v8.1".
         If omitted, the output files will not be labelled with the version number, by default None
@@ -199,6 +210,10 @@ def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, ou
         os.makedirs(output_directory)
     # master dataset file
     master_fp = os.path.join(output_directory, f"{domain_version_string}.xlsx")
+
+    ds_an_ls_df = stixToDf.detectionStrategiesAnalyticsLogSourcesDf(src)
+    add_ds_an_ls_to = {"detectionstrategies", "analytics", "datacomponents"}
+
     with pd.ExcelWriter(path=master_fp, engine="xlsxwriter") as master_writer:
         # master list of citations
         citations = pd.DataFrame()
@@ -217,6 +232,10 @@ def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, ou
                     for sheet_name in object_data:
                         logger.debug(f"Writing sheet to {fp}: {sheet_name}")
                         object_data[sheet_name].to_excel(object_writer, sheet_name=sheet_name, index=False)
+   
+                    # Write Detection strategy - Analytics - Log sources file
+                    if object_type in add_ds_an_ls_to and isinstance(ds_an_ls_df, pd.DataFrame) and not ds_an_ls_df.empty:
+                        ds_an_ls_df.to_excel(object_writer, sheet_name="detection mappings", index=False)
                 written_files.append(fp)
 
                 # add citations to master citations list
@@ -303,6 +322,8 @@ def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, ou
 
                 written_files.append(fp)
 
+        if isinstance(ds_an_ls_df, pd.DataFrame) and not ds_an_ls_df.empty:
+            ds_an_ls_df.to_excel(master_writer, sheet_name="detection mappings", index=False)
         # remove duplicate citations and add sheet to master file
         logger.debug(f"Writing sheet to {master_fp}: citations")
         citations.drop_duplicates(subset="reference", ignore_index=True).sort_values("reference").to_excel(
@@ -311,17 +332,6 @@ def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, ou
 
     written_files.append(master_fp)
 
-    # Write Detection strategy - Analytics - Log sources file
-    ds_an_lg_frames = build_ds_an_lg_relationships(dataframes)
-    ds_an_lg_fp = os.path.join(output_directory, f"{domain_version_string}-detectionstrategy-analytic-logsources.xlsx")
-
-    with pd.ExcelWriter(ds_an_lg_fp) as rel_writer:
-        for sheet_name, df in ds_an_lg_frames.items():
-            if not df.empty:
-                df.to_excel(rel_writer, sheet_name=sheet_name, index=False)
-
-    written_files.append(ds_an_lg_fp)
-
     for thefile in written_files:
         logger.info(f"Excel file created: {thefile}")
     return written_files
@@ -398,7 +408,7 @@ def export(
                 return
 
     dataframes = build_dataframes(src=mem_store, domain=domain)
-    write_excel(dataframes=dataframes, domain=domain, version=version, output_dir=output_dir)
+    write_excel(dataframes=dataframes, domain=domain, src=mem_store, version=version, output_dir=output_dir)
 
 
 def main():
@@ -410,7 +420,7 @@ def main():
         "-domain",
         type=str,
         choices=["enterprise-attack", "mobile-attack", "ics-attack"],
-        default="enterprise-attack",
+        default="ics-attack",
         help="which domain of ATT&CK to convert",
     )
     parser.add_argument(

mitreattack/attackToExcel/stixToDf.py

@@ -382,7 +382,7 @@ def analyticsToDf(src):
         for ds in detection_strategies:
             for analytic_id in ds.get("x_mitre_analytic_refs", []):
                 analytic_to_ds_map.setdefault(analytic_id, []).append(
-                    {"detection_strategy_id": ds["id"], "detection_strategy_name": ds.get("name", "")}
+                    {"detection_strategy_attack_id": ds["external_references"][0]["external_id"], "detection_strategy_id": ds["id"], "detection_strategy_name": ds.get("name", "")}
                 )
 
         for analytic in tqdm(analytics, desc="parsing analytics"):
@@ -393,15 +393,18 @@ def analyticsToDf(src):
                 data_comp_id = logsrc.get("x_mitre_data_component_ref", "")
                 data_comp = src.get(data_comp_id)
                 data_comp_name = data_comp.get("name", "") if data_comp else ""
+                data_comp_attack_id = data_comp["external_references"][0]["external_id"]
 
                 logsource_rows.append(
                     {
                         "analytic_id": analytic["id"],
                         "analytic_name": analytic["external_references"][0]["external_id"],
                         "data_component_id": data_comp_id,
                         "data_component_name": data_comp_name,
+                        "data_component_attack_id": data_comp_attack_id,
                         "log_source_name": logsrc.get("name", ""),
                         "channel": logsrc.get("channel", ""),
+                        "platforms": ", ".join(sorted(analytic.get("x_mitre_platforms", [])))
                     }
                 )
 
@@ -412,13 +415,14 @@ def analyticsToDf(src):
                         "analytic_id": analytic["id"],
                         "analytic_name": analytic["external_references"][0]["external_id"],
                         "detection_strategy_id": ds_info["detection_strategy_id"],
+                        "detection_strategy_attack_id": ds_info["detection_strategy_attack_id"],
                         "detection_strategy_name": ds_info["detection_strategy_name"],
+                        "platforms": ", ".join(sorted(analytic.get("x_mitre_platforms", [])))
+
                     }
                 )
 
         dataframes["analytics"] = pd.DataFrame(analytic_rows).sort_values("name")
-        dataframes["analytic-logsource"] = pd.DataFrame(logsource_rows)
-        dataframes["analytic-detectionstrategy"] = pd.DataFrame(analytic_to_ds_rows)
 
         citations = get_citations(analytics)
         if not citations.empty:
@@ -454,18 +458,19 @@ def detectionstrategiesToDf(src):
 
                 rel_rows.append(
                     {
+                        "detection_strategy_attack_id": detection_strategy["external_references"][0]["external_id"],
                         "detection_strategy_id": detection_strategy["id"],
                         "detection_strategy_name": detection_strategy.get("name", ""),
                         "analytic_id": analytic_id,
                         "analytic_name": analytic_obj["external_references"][0]["external_id"],
+                        "platforms": ", ".join(sorted(analytic_obj.get("x_mitre_platforms", [])))
+
                     }
                 )
 
         # Build main dataframes
         dataframes["detectionstrategies"] = pd.DataFrame(detection_strategy_rows).sort_values("name")
 
-        dataframes["detectionstrategies-analytic"] = pd.DataFrame(rel_rows)
-
         citations = get_citations(detection_strategies)
         if not citations.empty:
             dataframes["citations"] = citations.sort_values("reference")
@@ -520,6 +525,46 @@ def softwareToDf(src):
 
     return dataframes
 
+def detectionStrategiesAnalyticsLogSourcesDf(src):
+    """Build a single DS -> LogSource -> Analytic dataframe directly from STIX."""
+    detection_strategies = src.query([Filter("type", "=", "x-mitre-detection-strategy")])
+    detection_strategies = remove_revoked_deprecated(detection_strategies)
+
+    analytics = src.query([Filter("type", "=", "x-mitre-analytic")])
+    analytics = remove_revoked_deprecated(analytics)
+    analytics_by_id = {a["id"]: a for a in analytics}
+
+    rows = []
+    for ds in detection_strategies:
+        ds_attack_id = ds.get("external_references", [{}])[0].get("external_id", "")
+        ds_id = ds.get("id", "")
+        ds_name = ds.get("name", "")
+
+        for analytic_id in ds.get("x_mitre_analytic_refs", []):
+            analytic = analytics_by_id.get(analytic_id)
+            analytic_attack_id = analytic["external_references"][0]["external_id"]
+            platforms = ", ".join(sorted(analytic.get("x_mitre_platforms", [])))
+
+            logsrc_refs = analytic.get("x_mitre_log_source_references", [])
+            for logsrc in logsrc_refs:
+                data_comp_id = logsrc.get("x_mitre_data_component_ref", "")
+                data_comp = src.get(data_comp_id)
+
+                rows.append({
+                    "detection_strategy_attack_id": ds_attack_id,
+                    "detection_strategy_id": ds_id,
+                    "detection_strategy_name": ds_name,
+                    "analytic_id": analytic_id,
+                    "analytic_name": analytic_attack_id,
+                    "platforms": platforms,
+                    "log_source_name": logsrc.get("name", ""),
+                    "channel": logsrc.get("channel", ""),
+                    "data_component_id": data_comp_id,
+                    "data_component_name": (data_comp.get("name", "") if data_comp else ""),
+                    "data_component_attack_id": data_comp["external_references"][0]["external_id"]
+                })
+
+    return pd.DataFrame(rows)
 
 def groupsToDf(src):
     """Parse STIX groups from the given data and return corresponding pandas dataframes.
@@ -1264,4 +1309,4 @@ def _get_relationship_citations(object_dataframe, relationship_df):
         else:
             for i in range(0, len(new_citations)):
                 new_citations[i] = ",".join([new_citations[i], subset[i]])
-    return new_citations
+    return new_citations