Bump some versions in pyproject.toml.

Restructure the Pydantic Object for
building MitreAttack API Groups
based on the newer version of attackcti library.

Author: pyth0n1c

contentctl/enrichments/attack_enrichment.py

@@ -33,27 +33,33 @@ def getEnrichmentByMitreID(self, mitre_id:Annotated[str, Field(pattern=r"^T\d{4}
         else:
             raise Exception(f"Error, Unable to find Mitre Enrichment for MitreID {mitre_id}")
 
-
-    def addMitreID(self, technique:dict, tactics:list[str], groups:list[dict[str,Any]])->None:
+    def addMitreIDViaGroupNames(self, technique:dict, tactics:list[str], groupNames:list[str])->None:
+        technique_id = technique['technique_id']
+        technique_obj = technique['technique']
+        tactics.sort()
 
+        if technique_id in self.data:
+            raise Exception(f"Error, trying to redefine MITRE ID '{technique_id}'")
+        self.data[technique_id] = MitreAttackEnrichment(mitre_attack_id=technique_id, 
+                                                        mitre_attack_technique=technique_obj, 
+                                                        mitre_attack_tactics=tactics, 
+                                                        mitre_attack_groups=groupNames,
+                                                        mitre_attack_group_objects=[]) 
+
+    def addMitreIDViaGroupObjects(self, technique:dict, tactics:list[str],  groupObjects:list[dict[str,Any]])->None:
         technique_id = technique['technique_id']
         technique_obj = technique['technique']
         tactics.sort()
-        group_names_only:list[str] = sorted([group['group'] for group in groups])
 
-        import pprint
-        print(technique_id)
-        print(technique_obj)
-        print(tactics)
-        print(group_names_only)
-        pprint.pprint(groups)
+        groupNames:list[str] = sorted([group['group'] for group in groupObjects])
+        
         if technique_id in self.data:
             raise Exception(f"Error, trying to redefine MITRE ID '{technique_id}'")
         self.data[technique_id] = MitreAttackEnrichment(mitre_attack_id=technique_id, 
                                                         mitre_attack_technique=technique_obj, 
                                                         mitre_attack_tactics=tactics, 
-                                                        mitre_attack_groups=group_names_only,
-                                                        mitre_attack_group_objects=groups)
+                                                        mitre_attack_groups=groupNames,
+                                                        mitre_attack_group_objects=groupObjects)
 
 
     def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cached_or_offline: bool = False, skip_enrichment:bool = False) -> dict:
@@ -105,7 +111,7 @@ def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cach
                     for tactic in technique['tactic']:
                         tactics.append(tactic.replace('-',' ').title())
 
-                self.addMitreID(technique, tactics, apt_groups)
+                self.addMitreIDViaGroupObjects(technique, tactics, apt_groups)
                 attack_lookup[technique['technique_id']] = {'technique': technique['technique'], 'tactics': tactics, 'groups': apt_groups}
 
             if store_csv:
@@ -138,7 +144,7 @@ def get_attack_lookup(self, input_path: str, store_csv: bool = False, force_cach
                 technique_input = {'technique_id': key , 'technique': attack_lookup[key]['technique'] }
                 tactics_input = attack_lookup[key]['tactics']
                 groups_input = attack_lookup[key]['groups']
-                self.addMitreID(technique=technique_input, tactics=tactics_input, groups=groups_input)
+                self.addMitreIDViaGroupNames(technique=technique_input, tactics=tactics_input, groups=groups_input)
 
 
 

contentctl/objects/mitre_attack_enrichment.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-from pydantic import BaseModel, Field, ConfigDict, HttpUrl
+from pydantic import BaseModel, Field, ConfigDict, HttpUrl, field_validator
 from typing import List, Annotated
 from enum import StrEnum
 import datetime
@@ -22,17 +22,14 @@ class MitreTactics(StrEnum):
 
 
 class AttackGroupMatrix(StrEnum):
-    mitre_attack = "mitre-attack"
+    enterprise_attack = "enterprise-attack"
+    ics_attack = "ics-attack"
+    mobile_attack = "mobile-attack"
 
 
 class AttackGroupType(StrEnum):
     intrusion_set = "intrusion-set"
 
-class MitreDomain(StrEnum):
-    intrusion_set = "enterprise-attack"
-    mobile_attack = "mobile-attack"
-    ics_attack = "ics-attack"
-
 class MitreExternalReference(BaseModel):
     model_config = ConfigDict(extra='forbid')
     source_name: str
@@ -43,25 +40,47 @@ class MitreExternalReference(BaseModel):
 
 class MitreAttackGroup(BaseModel):
     model_config = ConfigDict(extra='forbid')
+    contributors: list[str] = []
     created: datetime.datetime
     created_by_ref: str
     external_references: list[MitreExternalReference]
     group: str
     group_aliases: list[str]
     group_description: str
+    group_id: str
     id: str
-    matrix: AttackGroupMatrix
+    matrix: list[AttackGroupMatrix]
+    mitre_attack_spec_version: None | str
+    mitre_version: str
+    #assume that if the deprecated field is not present, then the group is not deprecated
+    mitre_deprecated: bool
     modified: datetime.datetime
+    modified_by_ref: str
     object_marking_refs: list[str]
     type: AttackGroupType
     url: HttpUrl
-    x_mitre_attack_spec_version: None | str = None
-    x_mitre_deprecated: None | bool = None
-    x_mitre_domains: list[MitreDomain]
-    x_mitre_modified_by_ref: str
-    x_mitre_version: str
-    contributors: list[str] = []
+    
+
+    @field_validator("mitre_deprecated", mode="before")
+    def standardize_mitre_deprecated(cls, mitre_deprecated:bool | None) -> bool:
+        '''
+        For some reason, the API will return either a bool for mitre_deprecated OR
+        None. We simplify our typing by converting None to False, and assuming that
+        if deprecated is None, then the group is not deprecated.
+        '''
+        if mitre_deprecated is None:
+            return False
+        return mitre_deprecated
 
+    @field_validator("contributors", mode="before")
+    def standardize_contributors(cls, contributors:list[str] | None) -> list[str]:
+        '''
+        For some reason, the API will return either a list of strings for contributors OR
+        None. We simplify our typing by converting None to an empty list.
+        '''
+        if contributors is None:
+            return []
+        return contributors
 
 class MitreAttackEnrichment(BaseModel):
     ConfigDict(use_enum_values=True)

pyproject.toml

@@ -11,20 +11,20 @@ contentctl = 'contentctl.contentctl:main'
 
 [tool.poetry.dependencies]
 python = "^3.11"
-pydantic = "^2.7.1"
-PyYAML = "^6.0.1"
-requests = "~2.32.2"
+pydantic = "^2.8.2"
+PyYAML = "^6.0.2"
+requests = "~2.32.3"
 pycvesearch = "^1.2"
 xmltodict = "^0.13.0"
-attackcti = ">=0.3.7,<0.5.0"
+attackcti = "^0.4.0"
 Jinja2 = "^3.1.4"
 questionary = "^2.0.1"
 docker = "^7.1.0"
-splunk-sdk = "^2.0.1"
+splunk-sdk = "^2.0.2"
 semantic-version = "^2.10.0"
 bottle = "^0.12.25"
-tqdm = "^4.66.4"
-pygit2 = "^1.14.1"
+tqdm = "^4.66.5"
+pygit2 = "^1.15.1"
 tyro = "^0.8.3"
 gitpython = "^3.1.43"
 setuptools = ">=69.5.1,<74.0.0"