diff --git a/api_app/classes.py b/api_app/classes.py index f347bcfd6d..109e5e8cba 100644 --- a/api_app/classes.py +++ b/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__, ) diff --git a/api_app/connectors_manager/classes.py b/api_app/connectors_manager/classes.py index 1dd46d98ea..a68159eb24 100644 --- a/api_app/connectors_manager/classes.py +++ b/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,135 @@ 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. + """ + + @abc.abstractmethod + def run(self) -> dict: + raise NotImplementedError() + + # ── 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 diff --git a/api_app/connectors_manager/connectors/misp.py b/api_app/connectors_manager/connectors/misp.py index 51ea8c63fb..445d147578 100644 --- a/api_app/connectors_manager/connectors/misp.py +++ b/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 diff --git a/api_app/connectors_manager/connectors/opencti.py b/api_app/connectors_manager/connectors/opencti.py index 7fd9cc1526..db88733dda 100644 --- a/api_app/connectors_manager/connectors/opencti.py +++ b/api_app/connectors_manager/connectors/opencti.py @@ -8,9 +8,8 @@ from django.conf import settings from pycti.api.opencti_api_client import File -from api_app import helpers from api_app.choices import Classification -from api_app.connectors_manager import classes +from api_app.connectors_manager.classes import CTIConnector logger = logging.getLogger(__name__) @@ -24,11 +23,11 @@ # type hash is missing because it is combined with "file" # "generic" is misc field, so keeping text Classification.GENERIC: "x-opencti-text", - "file": "file", # hashes: md5, sha-1, sha-256 + Classification.FILE: "file", # hashes: md5, sha-1, sha-256 } -class OpenCTI(classes.Connector): +class OpenCTI(CTIConnector): ssl_verify: bool tlp: dict proxies: str @@ -36,22 +35,20 @@ class OpenCTI(classes.Connector): _api_key_name: str def get_observable_type(self) -> str: - if self._job.is_sample: - obs_type = INTELOWL_OPENCTI_TYPE_MAP["file"] - elif self._job.analyzable.classification == Classification.HASH: - matched_hash_type = helpers.get_hash_type(self._job.analyzable.name) - if matched_hash_type in [ + if self.classification == Classification.FILE: + obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.FILE] + elif self.classification == Classification.HASH: + if self.hash_type in [ "md5", "sha-1", "sha-256", ]: # sha-512 not supported - obs_type = INTELOWL_OPENCTI_TYPE_MAP["file"] + obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.FILE] else: obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.GENERIC] # text - elif self._job.analyzable.classification == Classification.IP: - ip_version = helpers.get_ip_version(self._job.analyzable.name) - if ip_version in [4, 6]: - obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.IP][f"v{ip_version}"] # v4/v6 + elif self.classification == Classification.IP: + if (ip_ver := self.ip_version) in (4, 6): + obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.IP][f"v{ip_ver}"] # v4/v6 else: obs_type = INTELOWL_OPENCTI_TYPE_MAP[Classification.GENERIC] # text else: @@ -62,18 +59,17 @@ def get_observable_type(self) -> str: def generate_observable_data(self) -> dict: observable_data = {"type": self.get_observable_type()} if self._job.is_sample: - observable_data["name"] = self._job.analyzable.name + observable_data["name"] = self.observable_name observable_data["hashes"] = { "md5": self._job.analyzable.md5, "sha-1": self._job.analyzable.sha1, "sha-256": self._job.analyzable.sha256, } - elif self._job.analyzable.classification == Classification.HASH and observable_data["type"] == "file": + elif self.classification == Classification.HASH and observable_data["type"] == Classification.FILE: # add hash instead of value - matched_type = helpers.get_hash_type(self._job.analyzable.name) - observable_data["hashes"] = {matched_type: self._job.analyzable.name} + observable_data["hashes"] = {self.hash_type: self.observable_name} else: - observable_data["value"] = self._job.analyzable.name + observable_data["value"] = self.observable_name return observable_data @@ -127,25 +123,55 @@ def _create_observable(self, created, org_id, marking_id): def _create_labels(self, created): label_ids = [] for tag in self._job.tags.all(): - label = pycti.Label(self.opencti_instance).create( + opencti_label = pycti.Label(self.opencti_instance).create( value=f"intelowl-tag:{tag.label}", color=tag.color, ) - if not isinstance(label, dict) or "id" not in label: + if not isinstance(opencti_label, dict) or "id" not in opencti_label: raise ValueError("Invalid response from OpenCTI Label.create") - label_id = label["id"] + label_id = opencti_label["id"] created["labels"].append(label_id) label_ids.append(label_id) + + # Add enrichment labels from data model when available + if self.has_data_model: + enrichment_labels = [] + if self.evaluation: + enrichment_labels.append(f"evaluation:{self.evaluation}") + if self.malware_family: + enrichment_labels.append(f"malware-family:{self.malware_family}") + if self.kill_chain_phase: + enrichment_labels.append(f"kill-chain:{self.kill_chain_phase}") + if self.reliability: + enrichment_labels.append(f"reliability:{self.reliability}") + + for enrichment_value in enrichment_labels: + enrichment_label = pycti.Label(self.opencti_instance).create( + value=enrichment_value, + ) + if isinstance(enrichment_label, dict) and "id" in enrichment_label: + label_id = enrichment_label["id"] + created["labels"].append(label_id) + label_ids.append(label_id) + return label_ids def _create_report(self, created, org_id, marking_id, label_ids): + description = ( + f"This is IntelOwl's analysis report for Job: {self.job_id}." + f" Analyzers Executed: {', '.join(self.analyzer_names)}" + ) + + # Append enrichment summary to report description when available + if self.has_data_model: + enrichment = self.get_enrichment_summary() + if enrichment: + enrichment_lines = [f" - {k}: {v}" for k, v in enrichment.items()] + description += "\n\nEnrichment Data:\n" + "\n".join(enrichment_lines) + report = pycti.Report(self.opencti_instance).create( name=f"IntelOwl Job-{self.job_id}", - description=( - f"This is IntelOwl's analysis report for Job: {self.job_id}." - " Analyzers Executed:" - f" {', '.join(list(self._job.analyzers_to_execute.all().values_list('name', flat=True)))}" - ), + description=description, published=self._job.received_request_time.strftime("%Y-%m-%dT%H:%M:%SZ"), report_types=["internal-report"], createdBy=org_id, @@ -164,7 +190,7 @@ def _create_external_reference(self, created): external_reference = pycti.ExternalReference(self.opencti_instance, None).create( source_name="IntelOwl Analysis", description="View analysis report on the IntelOwl instance", - url=f"{settings.WEB_CLIENT_URL}/jobs/{self.job_id}", + url=self.analysis_url, ) if not isinstance(external_reference, dict) or "id" not in external_reference: created["external_reference"] = None diff --git a/api_app/connectors_manager/connectors/yeti.py b/api_app/connectors_manager/connectors/yeti.py index c03baf0919..153933eda3 100644 --- a/api_app/connectors_manager/connectors/yeti.py +++ b/api_app/connectors_manager/connectors/yeti.py @@ -1,19 +1,18 @@ # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl # See the file 'LICENSE' for copying permission. -import ipaddress import logging import requests from django.conf import settings -from api_app.connectors_manager import classes +from api_app.connectors_manager.classes import CTIConnector from api_app.connectors_manager.exceptions import ConnectorRunException logger = logging.getLogger(__name__) -class YETI(classes.Connector): +class YETI(CTIConnector): verify_ssl: bool _url_key_name: str _api_key_name: str @@ -75,47 +74,49 @@ def health_check(self, user=None) -> tuple: logger.exception(f"Unexpected error in YETI health_check: {e}") return False, f"Unexpected error: {e}" - def run(self): - # get observable value and type - if self._job.is_sample: - obs_value = self._job.analyzable.md5 - obs_type = "file" + def _get_yeti_observable_type(self) -> str: + """ + Convert IntelOwl classification to YETI's expected observable type. + """ + obs_classification = self.classification + + if obs_classification == "ip": + ip_ver = self.ip_version + if ip_ver == 4: + return "ipv4" + elif ip_ver == 6: + return "ipv6" + else: + return "generic" + elif obs_classification == "domain": + return "hostname" + elif obs_classification == "hash": + return "generic" else: - obs_value = self._job.analyzable.name - obs_type = self._job.analyzable.classification - - # convert obs_type to YETI's expected types if possible - if obs_type == "ip": - # mark whether the IP is ipv4 or ipv6, fallback to generic on error - try: - ip_obj = ipaddress.ip_address(obs_value) - if ip_obj.version == 4: - obs_type = "ipv4" - elif ip_obj.version == 6: - obs_type = "ipv6" - else: - obs_type = "generic" - except Exception: - obs_type = "generic" - elif obs_type == "domain": - obs_type = "hostname" - elif obs_type == "hash": - obs_type = "generic" + return obs_classification + + def run(self): + obs_value = self.observable_value + obs_type = self._get_yeti_observable_type() # create context context = { "source": "IntelOwl", - "report": f"{settings.WEB_CLIENT_URL}/jobs/{self.job_id}", + "report": self.analysis_url, "status": "analyzed", "date": str(self._job.received_request_time), "description": f"IntelOwl's analysis report for Job: {self.job_id} | {obs_value} | {obs_type}", - "analyzers executed": ", ".join( - list(self._job.analyzers_to_execute.all().values_list("name", flat=True)) - ), + "analyzers executed": ", ".join(self.analyzer_names), } + # Add enrichment data to context when available + if self.has_data_model: + enrichment = self.get_enrichment_summary() + for key, value in enrichment.items(): + context[key] = str(value) if not isinstance(value, str) else value + # get job tags - tags = list(self._job.tags.all().values_list("label", flat=True)) + tags = self.tag_labels # request payload payload = { diff --git a/tests/api_app/connectors_manager/test_classes.py b/tests/api_app/connectors_manager/test_classes.py index a617c0f851..ae78423674 100644 --- a/tests/api_app/connectors_manager/test_classes.py +++ b/tests/api_app/connectors_manager/test_classes.py @@ -3,6 +3,7 @@ from unittest.mock import patch +from django.conf import settings from kombu import uuid from api_app.analyzables_manager.models import Analyzable @@ -155,3 +156,257 @@ def run(self) -> dict: cc.delete() job.delete() an.delete() + + +class CTIConnectorTestCase(CustomTestCase): + fixtures = [ + "api_app/fixtures/0001_user.json", + ] + + @staticmethod + def _create_job(name, classification): + an = Analyzable.objects.create( + name=name, + classification=classification, + ) + job = Job.objects.create( + analyzable=an, + status=Job.STATUSES.CONNECTORS_RUNNING.value, + ) + return job, an + + @staticmethod + def _create_cti_connector(job): + from api_app.connectors_manager.classes import CTIConnector + + pm = PythonModule.objects.get(base_path=PythonModuleBasePaths.Connector.value, module="misp.MISP") + cc = ConnectorConfig.objects.create( + name="test_cti", + python_module=pm, + description="test cti connector", + disabled=True, + maximum_tlp="CLEAR", + ) + + class MockCTIConnector(CTIConnector): + def run(self) -> dict: + return {} + + connector = MockCTIConnector(cc) + connector.job_id = job.pk + return connector, cc + + def test_observable_name_domain(self): + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.observable_name, "example.com") + + cc.delete() + job.delete() + an.delete() + + def test_observable_value_observable(self): + job, an = self._create_job("8.8.8.8", Classification.IP) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.observable_value, "8.8.8.8") + + cc.delete() + job.delete() + an.delete() + + def test_classification_ip(self): + job, an = self._create_job("8.8.8.8", Classification.IP) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.classification, Classification.IP) + + cc.delete() + job.delete() + an.delete() + + def test_classification_domain(self): + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.classification, Classification.DOMAIN) + + cc.delete() + job.delete() + an.delete() + + def test_hash_type_md5(self): + md5_hash = "d" * 32 + job, an = self._create_job(md5_hash, Classification.HASH) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.hash_type, "md5") + + cc.delete() + job.delete() + an.delete() + + def test_hash_type_sha256(self): + sha256_hash = "a" * 64 + job, an = self._create_job(sha256_hash, Classification.HASH) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.hash_type, "sha-256") + + cc.delete() + job.delete() + an.delete() + + def test_hash_type_none_for_non_hash(self): + job, an = self._create_job("8.8.8.8", Classification.IP) + connector, cc = self._create_cti_connector(job) + + self.assertIsNone(connector.hash_type) + + cc.delete() + job.delete() + an.delete() + + def test_ip_version_v4(self): + job, an = self._create_job("8.8.8.8", Classification.IP) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.ip_version, 4) + + cc.delete() + job.delete() + an.delete() + + def test_ip_version_v6(self): + job, an = self._create_job("::1", Classification.IP) + connector, cc = self._create_cti_connector(job) + + self.assertEqual(connector.ip_version, 6) + + cc.delete() + job.delete() + an.delete() + + def test_ip_version_none_for_non_ip(self): + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + self.assertIsNone(connector.ip_version) + + cc.delete() + job.delete() + an.delete() + + def test_analysis_url(self): + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + expected = f"{settings.WEB_CLIENT_URL}/jobs/{job.pk}" + self.assertEqual(connector.analysis_url, expected) + + cc.delete() + job.delete() + an.delete() + + def test_tag_labels(self): + from api_app.models import Tag + + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + tag1 = Tag.objects.create(label="test-tag-1", color="#FF0000") + tag2 = Tag.objects.create(label="test-tag-2", color="#00FF00") + job.tags.add(tag1, tag2) + + labels = connector.tag_labels + self.assertIn("test-tag-1", labels) + self.assertIn("test-tag-2", labels) + self.assertEqual(len(labels), 2) + + job.tags.clear() + tag1.delete() + tag2.delete() + cc.delete() + job.delete() + an.delete() + + def test_enrichment_without_data_model(self): + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + # Job has no data_model set -- all enrichment should be None/empty + self.assertFalse(connector.has_data_model) + self.assertIsNone(connector.evaluation) + self.assertIsNone(connector.malware_family) + self.assertIsNone(connector.kill_chain_phase) + self.assertIsNone(connector.reliability) + self.assertEqual(connector.related_threats, []) + self.assertEqual(connector.data_model_tags, []) + self.assertEqual(connector.external_references, []) + self.assertEqual(connector.get_enrichment_summary(), {}) + + cc.delete() + job.delete() + an.delete() + + def test_enrichment_with_data_model(self): + from api_app.data_model_manager.models import DomainDataModel + + job, an = self._create_job("example.com", Classification.DOMAIN) + connector, cc = self._create_cti_connector(job) + + # Create a real data model and attach to job + dm = DomainDataModel.objects.create( + evaluation="malicious", + malware_family="emotet", + kill_chain_phase="delivery", + reliability=8, + related_threats=["threat1", "threat2"], + tags=["malware", "phishing"], + external_references=["https://example.com/report"], + ) + job.data_model = dm + job.save() + + # Clear the cached_property so it re-reads + if "_merged_data_model" in connector.__dict__: + del connector.__dict__["_merged_data_model"] + # Also clear the cached _job property + if "_job" in connector.__dict__: + del connector.__dict__["_job"] + + self.assertTrue(connector.has_data_model) + self.assertEqual(connector.evaluation, "malicious") + self.assertEqual(connector.malware_family, "emotet") + self.assertEqual(connector.kill_chain_phase, "delivery") + self.assertEqual(connector.reliability, 8) + self.assertIn("threat1", connector.related_threats) + self.assertIn("threat2", connector.related_threats) + self.assertIn("malware", connector.data_model_tags) + self.assertIn("phishing", connector.data_model_tags) + self.assertIn("https://example.com/report", connector.external_references) + + summary = connector.get_enrichment_summary() + self.assertEqual(summary["evaluation"], "malicious") + self.assertEqual(summary["malware_family"], "emotet") + self.assertEqual(summary["kill_chain_phase"], "delivery") + self.assertEqual(summary["reliability"], 8) + + dm.delete() + cc.delete() + job.delete() + an.delete() + + def test_enrichment_generic_classification(self): + # Generic classification has no data model, enrichment should be None/empty + job, an = self._create_job("some-random-text", Classification.GENERIC) + connector, cc = self._create_cti_connector(job) + + self.assertFalse(connector.has_data_model) + self.assertIsNone(connector.evaluation) + self.assertEqual(connector.get_enrichment_summary(), {}) + + cc.delete() + job.delete() + an.delete() diff --git a/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py b/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py index 2a5031fe08..4802dc0c62 100644 --- a/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py +++ b/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py @@ -65,10 +65,8 @@ def get_extra_config(cls) -> dict: def _setup_connector(self): connector = super()._setup_connector() - mock_tag = MagicMock() - mock_tag.label = "source:intelowl" connector._job.tags = MagicMock() - connector._job.tags.all.return_value = [mock_tag] + connector._job.tags.all.return_value.values_list.return_value = ["source:intelowl"] connector._job.analyzers_to_execute = MagicMock() connector._job.analyzers_to_execute.all.return_value.values_list.return_value = ["FireHol_IPList"]