[Connectors] Introduce CTIConnector to standardize CTI connectors and support data model enrichment

api_app/classes.py

@@ -74,7 +74,12 @@ def all_subclasses(cls):
             __import__(package)
         classes = cls.__subclasses__()
         return sorted(
-            [class_ for class_ in classes if not class_.__name__.startswith("MockUp")],
+            [
+                class_
+                for class_ in classes
+                if not class_.__name__.startswith("MockUp")
+                and class_.__module__.startswith(str(cls.python_base_path))
+            ],
             key=lambda x: x.__name__,
         )
 

api_app/connectors_manager/classes.py

@@ -2,8 +2,14 @@
 # See the file 'LICENSE' for copying permission.
 import abc
 import logging
-from typing import Type
+import typing
+from typing import List, Optional, Type
 
+from django.conf import settings
+from django.utils.functional import cached_property
+
+from api_app import helpers
+from api_app.choices import Classification
 from api_app.decorators import classproperty
 
 from ..choices import PythonModuleBasePaths, ReportStatus
@@ -66,3 +72,131 @@ def before_run(self):
     def after_run(self):
         super().after_run()
         logger.info(f"FINISHED connector: {self.__repr__()}")
+
+
+class CTIConnector(Connector):
+    """
+    Base class for Cyber Threat Intelligence connectors (MISP, OpenCTI, YETI).
+
+    Provides standardized, reusable properties for:
+    - Observable metadata extraction (name, value, classification, hash type, IP version)
+    - Job context (analysis URL, tag labels, analyzer names)
+    - Data-model enrichment (evaluation verdict, malware family, kill chain phase,
+      reliability score, related threats, etc.)
+
+    Subclasses should inherit from this class.
+    Enrichment properties are opt-in: each subclass decides which fields to include
+    in its platform-specific payload.
+    """
+
+    # ── Observable Metadata ──────────────────────────────────────
+
+    @property
+    def observable_name(self) -> str:
+        return self._job.analyzable.name
+
+    @property
+    def observable_value(self) -> str:
+        if self._job.is_sample:
+            return self._job.analyzable.md5
+        return self._job.analyzable.name
+
+    @property
+    def classification(self) -> str:
+        if self._job.is_sample:
+            return Classification.FILE
+        return self._job.analyzable.classification
+
+    @property
+    def hash_type(self) -> Optional[str]:
+        if not self._job.is_sample and self._job.analyzable.classification == Classification.HASH:
+            return helpers.get_hash_type(self._job.analyzable.name)
+        return None
+
+    @property
+    def ip_version(self) -> Optional[int]:
+        if not self._job.is_sample and self._job.analyzable.classification == Classification.IP:
+            return helpers.get_ip_version(self._job.analyzable.name)
+        return None
+
+    @property
+    def analysis_url(self) -> str:
+        return f"{settings.WEB_CLIENT_URL}/jobs/{self.job_id}"
+
+    @property
+    def tag_labels(self) -> List[str]:
+        return list(self._job.tags.all().values_list("label", flat=True))
+
+    @property
+    def analyzer_names(self) -> List[str]:
+        return list(self._job.analyzers_to_execute.all().values_list("name", flat=True))
+
+    # Data-Model Enrichment
+
+    @cached_property
+    def _merged_data_model(self) -> typing.Any:
+        # Returns the merged data model from the
+        # engine step, or None if unavailable
+
+        return getattr(self._job, "data_model", None)
+
+    @property
+    def has_data_model(self) -> bool:
+        return self._merged_data_model is not None
+
+    @property
+    def evaluation(self) -> Optional[str]:
+        # Verdict: trusted, malicious, or None.
+        dm = self._merged_data_model
+        return dm.evaluation if dm else None
+
+    @property
+    def malware_family(self) -> Optional[str]:
+        dm = self._merged_data_model
+        return dm.malware_family if dm else None
+
+    @property
+    def kill_chain_phase(self) -> Optional[str]:
+        dm = self._merged_data_model
+        return dm.kill_chain_phase if dm else None
+
+    @property
+    def reliability(self) -> Optional[int]:
+        dm = self._merged_data_model
+        return dm.reliability if dm else None
+
+    @property
+    def related_threats(self) -> List[str]:
+        dm = self._merged_data_model
+        return list(dm.related_threats) if dm and dm.related_threats else []
+
+    @property
+    def data_model_tags(self) -> List[str]:
+        dm = self._merged_data_model
+        return list(dm.tags) if dm and dm.tags else []
+
+    @property
+    def external_references(self) -> List[str]:
+        dm = self._merged_data_model
+        return list(dm.external_references) if dm and dm.external_references else []
+
+    def get_enrichment_summary(self) -> dict:
+        """
+        Returns a dict summarizing all available enrichment fields.
+        Only includes fields that have non-None/non-empty values.
+        Useful for connectors that want to bulk-attach enrichment metadata.
+        """
+        summary = {}
+        fields = {
+            "evaluation": self.evaluation,
+            "malware_family": self.malware_family,
+            "kill_chain_phase": self.kill_chain_phase,
+            "reliability": self.reliability,
+            "related_threats": self.related_threats,
+            "data_model_tags": self.data_model_tags,
+            "external_references": self.external_references,
+        }
+        for key, value in fields.items():
+            if value:
+                summary[key] = value
+        return summary

api_app/connectors_manager/connectors/misp.py

@@ -7,9 +7,8 @@
 import pymisp
 from django.conf import settings
 
-from api_app import helpers
 from api_app.choices import Classification
-from api_app.connectors_manager.classes import Connector
+from api_app.connectors_manager.classes import CTIConnector
 from api_app.connectors_manager.exceptions import ConnectorRunException
 
 logger = logging.getLogger(__name__)
@@ -18,9 +17,9 @@
     Classification.IP: "ip-src",
     Classification.DOMAIN: "domain",
     Classification.URL: "url",
-    # "hash" (checked from helpers.get_hash_type)
+    # "hash" (checked from self.hash_type)
     Classification.GENERIC: "text",  # misc field, so keeping text
-    "file": "filename|md5",
+    Classification.FILE: "filename|md5",
 }
 
 
@@ -31,7 +30,7 @@ def create_misp_attribute(misp_type, misp_value) -> pymisp.MISPAttribute:
     return obj
 
 
-class MISP(Connector):
+class MISP(CTIConnector):
     tlp: str
     ssl_check: bool
     self_signed_certificate: str
@@ -50,29 +49,37 @@ def _event_obj(self) -> pymisp.MISPEvent:
         obj.add_tag(f"tlp:{self.tlp}")  # tlp tag for sharing
 
         # Add tags from Job
-        for tag in self._job.tags.all():
-            obj.add_tag(f"intelowl-tag:{tag.label}")
+        for label in self.tag_labels:
+            obj.add_tag(f"intelowl-tag:{label}")
+
+        # Add enrichment tags from data model when available
+        if self.has_data_model:
+            if self.evaluation:
+                obj.add_tag(f"evaluation:{self.evaluation}")
+            if self.malware_family:
+                obj.add_tag(f"malware-family:{self.malware_family}")
+            if self.kill_chain_phase:
+                obj.add_tag(f"kill-chain:{self.kill_chain_phase}")
+            if self.reliability:
+                obj.add_tag(f"reliability:{self.reliability}")
 
         return obj
 
     @property
     def _base_attr_obj(self) -> pymisp.MISPAttribute:
-        if self._job.is_sample:
-            _type = INTELOWL_MISP_TYPE_MAP["file"]
-            value = f"{self._job.analyzable.name}|{self._job.analyzable.md5}"
+        if self.classification == Classification.FILE:
+            _type = INTELOWL_MISP_TYPE_MAP[Classification.FILE]
+            value = f"{self.observable_name}|{self._job.analyzable.md5}"
         else:
-            _type = self._job.analyzable.classification
-            value = self._job.analyzable.name
-            if _type == Classification.HASH:
-                matched_type = helpers.get_hash_type(value)
+            value = self.observable_name
+            if self.hash_type is not None:
                 # convert sha-x to shax
-                _type = matched_type.replace("-", "") if matched_type is not None else "text"
+                _type = self.hash_type.replace("-", "")
             else:
-                _type = INTELOWL_MISP_TYPE_MAP[_type]
+                _type = INTELOWL_MISP_TYPE_MAP.get(self.classification, "text")
 
         obj = create_misp_attribute(_type, value)
-        analyzers_names = self._job.analyzers_to_execute.all().values_list("name", flat=True)
-        obj.comment = f"Analyzers Executed: {', '.join(analyzers_names)}"
+        obj.comment = f"Analyzers Executed: {', '.join(self.analyzer_names)}"
         return obj
 
     @property
@@ -90,7 +97,7 @@ def _link_attr_obj(self) -> pymisp.MISPAttribute:
         """
         obj = pymisp.MISPAttribute()
         obj.type = "link"
-        obj.value = f"{settings.WEB_CLIENT_URL}/jobs/{self.job_id}"
+        obj.value = self.analysis_url
         obj.comment = "View Analysis on IntelOwl"
         obj.disable_correlation = True