HuntingAbuseAPI analyzer

api_app/analyzers_manager/migrations/0158_analyzer_config_huntingabuseapi.py

@@ -0,0 +1,141 @@
+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": "hunting_abuse.HuntingAbuseAPI",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "HuntingAbuseAPI",
+    "description": "[HuntingAbuseAPI](https://hunting.abuse.ch/api/) provides an updated list of false positives from all it's services. This API can be queried to verify. if the provided observable is valid or false positive.",
+    "disabled": False,
+    "soft_time_limit": 60,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "GREEN",
+    "observable_supported": ["ip", "url", "domain", "hash"],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "mapping_data_model": {
+        "$trusted": "evaluation",
+        "$9": "reliability",
+        "details": "additional_info",
+    },
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = [
+    {
+        "python_module": {
+            "module": "hunting_abuse.HuntingAbuseAPI",
+            "base_path": "api_app.analyzers_manager.observable_analyzers",
+        },
+        "name": "service_api_key",
+        "type": "str",
+        "description": "Mandatory API key to connect to abuse.ch services.",
+        "is_secret": True,
+        "required": True,
+    }
+]
+
+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]
+        # it is just the primary key serialized
+        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")
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    if not Model.objects.filter(name=plugin["name"]).exists():
+        exists = _create_object(Model, plugin)
+        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):
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    Model.objects.get(name=plugin["name"]).delete()
+
+
+class Migration(migrations.Migration):
+    atomic = False
+    dependencies = [
+        ("api_app", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0157_analyzer_config_phunter"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/observable_analyzers/hunting_abuse.py

@@ -0,0 +1,112 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+import json
+import logging
+import os
+
+import requests
+from django.conf import settings
+
+from api_app.analyzers_manager.classes import ObservableAnalyzer
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
+from api_app.choices import Classification
+from api_app.mixins import AbuseCHMixin
+from api_app.models import PluginConfig
+from tests.mock_utils import MockUpResponse, if_mock_connections, patch
+
+logger = logging.getLogger(__name__)
+
+db_name = "hunting_abuse_fplist.json"
+database_location = f"{settings.MEDIA_ROOT}/{db_name}"
+
+
+class HuntingAbuseAPI(AbuseCHMixin, ObservableAnalyzer):
+    url: str = "https://hunting-api.abuse.ch/api/v1/"
+
+    def _do_create_data_model(self) -> bool:
+        return super()._do_create_data_model() and self.report.report.get(
+            "fp_status", False
+        )
+
+    @classmethod
+    def get_auth_key(cls) -> str:
+        for plugin in PluginConfig.objects.filter(
+            parameter__python_module=cls.python_module,
+            parameter__is_secret=True,
+            parameter__name="service_api_key",
+        ):
+            if plugin.value:
+                return plugin.value
+        return ""
+
+    @classmethod
+    def update(cls) -> bool:
+        headers = {"Content-Type": "application/json", "Auth-Key": cls.get_auth_key()}
+        data = {"query": "get_fplist", "format": "json"}
+
+        try:
+            response = requests.post(cls.url, json=data, headers=headers)
+            response.raise_for_status()
+
+            with open(database_location, "w", encoding="utf-8") as f:
+                json.dump(response.json(), f)
+
+            if not os.path.exists(database_location):
+                raise FileNotFoundError(
+                    f"Failed to create the database file at {database_location}"
+                )
+            return True
+
+        except requests.RequestException as e:
+            logger.error(f"Failed to fetch response from Hunting Abuse API: {e}")
+        except FileNotFoundError as e:
+            logger.error(f"File not found: {e}")
+        except Exception as e:
+            logger.error(f"Failed to update Hunting Abuse database: {e}")
+
+        return False
+
+    def run(self):
+        if not os.path.isfile(database_location):
+            logger.info("Hunting Abuse database not found, updating...")
+            if not self.update():
+                raise AnalyzerRunException("Failed extraction of Hunting Abuse db")
+
+        with open(database_location, "r", encoding="utf-8") as f:
+            fp_list = json.load(f)
+
+        for _key, value_dict in fp_list.items():
+            # Checking observable classification is IP, is necessary to handle cases where response contains
+            # results in 'ip:port' format
+            is_match = self.observable_name == value_dict["entry_value"] or (
+                self.observable_classification == Classification.IP
+                and self.observable_name in value_dict["entry_value"]
+            )
+
+            if is_match:
+                return {"fp_status": True, "details": value_dict}
+        return {"fp_status": False}
+
+    @classmethod
+    def _monkeypatch(cls):
+        mock_response = {
+            "1": {
+                "time_stamp": "2025-06-04 07:46:14 UTC",
+                "platform": "MalwareBazaar",
+                "entry_type": "sha1_hash",
+                "entry_value": "ac4cb655a78a5634f6a87c82bec33a4391269a3f",
+                "removed_by": "admin",
+                "removal_notes": None,
+            }
+        }
+        patches = [
+            if_mock_connections(
+                patch(
+                    "requests.post",
+                    return_value=MockUpResponse(mock_response, 200),
+                ),
+            )
+        ]
+        return super()._monkeypatch(patches)