diff --git a/api_app/analyzers_manager/migrations/0197_analyzer_config_ismalicious.py b/api_app/analyzers_manager/migrations/0197_analyzer_config_ismalicious.py new file mode 100644 index 0000000000..35e52b4658 --- /dev/null +++ b/api_app/analyzers_manager/migrations/0197_analyzer_config_ismalicious.py @@ -0,0 +1,154 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from django.db import migrations +from django.db.models.fields.related_descriptors import ( + ForwardManyToOneDescriptor, + ForwardOneToOneDescriptor, + ManyToManyDescriptor, + ReverseManyToOneDescriptor, + ReverseOneToOneDescriptor, +) + +plugin = { + "python_module": { + "health_check_schedule": None, + "update_schedule": None, + "module": "ismalicious.IsMalicious", + "base_path": "api_app.analyzers_manager.observable_analyzers", + }, + "name": "IsMalicious", + "description": "Enrich IPs, domains, and URLs with [isMalicious](https://ismalicious.com) " + "threat intelligence: malicious flag, risk score, categories, and detection sources " + "via `GET /check`. Requires an API key from the isMalicious dashboard.", + "disabled": False, + "soft_time_limit": 30, + "routing_key": "default", + "health_check_status": True, + "type": "observable", + "docker_based": False, + "maximum_tlp": "AMBER", + "observable_supported": ["ip", "domain", "url"], + "supported_filetypes": [], + "run_hash": False, + "run_hash_type": "", + "not_supported_filetypes": [], + "mapping_data_model": {}, + "model": "analyzers_manager.AnalyzerConfig", +} + +params = [ + { + "python_module": { + "module": "ismalicious.IsMalicious", + "base_path": "api_app.analyzers_manager.observable_analyzers", + }, + "name": "api_key_name", + "type": "str", + "description": "isMalicious API credential (X-API-KEY). Dashboard → Account → Team Management.", + "is_secret": True, + "required": True, + }, + { + "python_module": { + "module": "ismalicious.IsMalicious", + "base_path": "api_app.analyzers_manager.observable_analyzers", + }, + "name": "url", + "type": "str", + "description": "isMalicious API base URL", + "is_secret": False, + "required": False, + }, +] + +values = [] + + +def _get_real_obj(Model, field, value): + def _get_obj(Model, other_model, value): + if isinstance(value, dict): + real_vals = {} + for key, real_val in value.items(): + real_vals[key] = _get_real_obj(other_model, key, real_val) + value = other_model.objects.get_or_create(**real_vals)[0] + else: + if isinstance(value, int): + if Model.__name__ == "PluginConfig": + value = other_model.objects.get(name=plugin["name"]) + else: + value = other_model.objects.get(pk=value) + else: + value = other_model.objects.get(name=value) + return value + + if ( + type(getattr(Model, field)) + in [ + ForwardManyToOneDescriptor, + ReverseManyToOneDescriptor, + ReverseOneToOneDescriptor, + ForwardOneToOneDescriptor, + ] + and value + ): + other_model = getattr(Model, field).get_queryset().model + value = _get_obj(Model, other_model, value) + elif type(getattr(Model, field)) in [ManyToManyDescriptor] and value: + other_model = getattr(Model, field).rel.model + value = [_get_obj(Model, other_model, val) for val in value] + return value + + +def _create_object(Model, data): + mtm, no_mtm = {}, {} + for field, value in data.items(): + value = _get_real_obj(Model, field, value) + if type(getattr(Model, field)) is ManyToManyDescriptor: + mtm[field] = value + else: + no_mtm[field] = value + try: + o = Model.objects.get(**no_mtm) + except Model.DoesNotExist: + o = Model(**no_mtm) + o.full_clean() + o.save() + for field, value in mtm.items(): + attribute = getattr(o, field) + if value is not None: + attribute.set(value) + return False + return True + + +def migrate(apps, schema_editor): + Parameter = apps.get_model("api_app", "Parameter") + PluginConfig = apps.get_model("api_app", "PluginConfig") + plugin_data = dict(plugin) + python_path = plugin_data.pop("model") + Model = apps.get_model(*python_path.split(".")) + if not Model.objects.filter(name=plugin_data["name"]).exists(): + exists = _create_object(Model, plugin_data) + if not exists: + for param in params: + _create_object(Parameter, param) + for value in values: + _create_object(PluginConfig, value) + + +def reverse_migrate(apps, schema_editor): + plugin_data = dict(plugin) + python_path = plugin_data.pop("model") + Model = apps.get_model(*python_path.split(".")) + Model.objects.get(name=plugin_data["name"]).delete() + + +class Migration(migrations.Migration): + atomic = False + dependencies = [ + ("api_app", "0073_alter_updatecheckstatus_last_checked_at_and_more"), + ("analyzers_manager", "0196_data_model_phishing_lists"), + ] + + operations = [migrations.RunPython(migrate, reverse_migrate)] diff --git a/api_app/analyzers_manager/observable_analyzers/ismalicious.py b/api_app/analyzers_manager/observable_analyzers/ismalicious.py new file mode 100644 index 0000000000..82771e485f --- /dev/null +++ b/api_app/analyzers_manager/observable_analyzers/ismalicious.py @@ -0,0 +1,59 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. +"""isMalicious observable analyzer for IntelOwl. + +Copy this file to: + api_app/analyzers_manager/observable_analyzers/ismalicious.py +""" + +from __future__ import annotations + +from typing import Any + +import requests + +from api_app.analyzers_manager.classes import ObservableAnalyzer +from api_app.analyzers_manager.exceptions import AnalyzerRunException + +DEFAULT_API_URL = "https://api.ismalicious.com" + + +def check_indicator( + query: str, + api_key: str, + api_url: str = DEFAULT_API_URL, + timeout: int = 30, +) -> dict[str, Any]: + """GET /check?query=…&enrichment=standard with X-API-KEY.""" + if not query or not query.strip(): + raise ValueError("query is required") + if not api_key: + raise ValueError("api_key is required") + + try: + response = requests.get( + f"{api_url.rstrip('/')}/check", + params={"query": query.strip(), "enrichment": "standard"}, + headers={"X-API-KEY": api_key, "Accept": "application/json"}, + timeout=timeout, + ) + response.raise_for_status() + except requests.RequestException as exc: + raise AnalyzerRunException(exc) from exc + return response.json() + + +class IsMalicious(ObservableAnalyzer): + url: str = DEFAULT_API_URL + _api_key_name: str + + @classmethod + def update(cls) -> bool: + return True + + def run(self) -> dict[str, Any]: + return check_indicator( + query=self.observable_name, + api_key=self._api_key_name, + api_url=self.url, + ) diff --git a/tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_ismalicious.py b/tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_ismalicious.py new file mode 100644 index 0000000000..eba3747e75 --- /dev/null +++ b/tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_ismalicious.py @@ -0,0 +1,33 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from unittest.mock import patch + +from api_app.analyzers_manager.observable_analyzers.ismalicious import IsMalicious +from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import ( + BaseAnalyzerTest, +) +from tests.mock_utils import MockUpResponse + +SAMPLE_REPORT = { + "malicious": True, + "riskScore": {"score": 80, "level": "high"}, + "categories": ["c2"], + "sources": [{"name": "feed-a"}], + "query": "8.8.8.8", +} + + +class IsMaliciousTestCase(BaseAnalyzerTest): + analyzer_class = IsMalicious + + @staticmethod + def get_mocked_response(): + return patch( + "api_app.analyzers_manager.observable_analyzers.ismalicious.requests.get", + return_value=MockUpResponse(SAMPLE_REPORT, 200), + ) + + @classmethod + def get_extra_config(cls) -> dict: + return {"_api_key_name": "fake_api_key"}