Add IsMalicious observable analyzer. Closes #3935

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)]

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,
+        )

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"}