<Analyzers> UltraDns, Closes #1783

Signed-off-by: pranjalg1331 <pranjaloff13@gmail.com>

api_app/analyzers_manager/migrations/0143_analyzer_config_ultradns_malicious_detector.py

@@ -0,0 +1,128 @@
+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": "dns.dns_malicious_detectors.ultradns_malicious_detector.UltraDNSMaliciousDetector",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "UltraDNS_Malicious_Detector",
+    "description": "Scan if a DNS is marked malicious by UltraDNS",
+    "disabled": False,
+    "soft_time_limit": 30,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "AMBER",
+    "observable_supported": ["url", "domain"],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "mapping_data_model": {},
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = []
+
+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", "0065_job_mpnodesearch"),
+        (
+            "analyzers_manager",
+            "0142_alter_analyzerreport_data_model_content_type_and_more",
+        ),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/migrations/0144_analyzer_config_ultradns_dns.py

@@ -0,0 +1,160 @@
+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": "dns.dns_resolvers.ultradns_dns_resolver.UltraDNSDNSResolver",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "UltraDNS_DNS",
+    "description": "Retrieve current domain resolution with UltraDNS",
+    "disabled": False,
+    "soft_time_limit": 30,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "AMBER",
+    "observable_supported": ["url", "domain"],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "mapping_data_model": {},
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = [
+    {
+        "python_module": {
+            "module": "dns.dns_resolvers.ultradns_dns_resolver.UltraDNSDNSResolver",
+            "base_path": "api_app.analyzers_manager.observable_analyzers",
+        },
+        "name": "query_type",
+        "type": "str",
+        "description": "",
+        "is_secret": False,
+        "required": False,
+    }
+]
+
+values = [
+    {
+        "parameter": {
+            "python_module": {
+                "module": "dns.dns_resolvers.ultradns_dns_resolver.UltraDNSDNSResolver",
+                "base_path": "api_app.analyzers_manager.observable_analyzers",
+            },
+            "name": "query_type",
+            "type": "str",
+            "description": "",
+            "is_secret": False,
+            "required": False,
+        },
+        "analyzer_config": "UltraDNS_DNS",
+        "connector_config": None,
+        "visualizer_config": None,
+        "ingestor_config": None,
+        "pivot_config": None,
+        "for_organization": False,
+        "value": "A",
+        "updated_at": "2024-12-25T11:31:43.211468Z",
+        "owner": None,
+    }
+]
+
+
+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", "0065_job_mpnodesearch"),
+        ("analyzers_manager", "0143_analyzer_config_ultradns_malicious_detector"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/ultradns_malicious_detector.py

@@ -0,0 +1,75 @@
+import ipaddress
+from urllib.parse import urlparse
+
+import dns.resolver
+
+from api_app.analyzers_manager import classes
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
+from tests.mock_utils import if_mock_connections, patch
+
+from ..dns_responses import malicious_detector_response
+
+
+class UltraDNSMaliciousDetector(classes.ObservableAnalyzer):
+    """Resolve a DNS query with UltraDNS servers,
+    if the response falls within the sinkhole range, the domain is malicious.
+    """
+
+    def update(self) -> bool:
+        pass
+
+    def run(self):
+        try:
+            is_malicious = False
+            observable = self.observable_name
+
+            # for URLs we are checking the relative domain
+            if self.observable_classification == self.ObservableTypes.URL:
+                observable = urlparse(self.observable_name).hostname
+
+            primary_dns = "156.154.70.2"
+            backup_dns = "156.154.71.2"
+            sinkhole_range = ipaddress.ip_network("156.154.112.0/23")
+
+            # Try primary DNS server first
+            resolver = dns.resolver.Resolver()
+            resolver.nameservers = [primary_dns]
+
+            try:
+                answers = resolver.resolve(observable, "A")
+                for rdata in answers:
+                    resolution = rdata.to_text()
+                    # Check if the resolution falls in the sinkhole range
+                    if ipaddress.ip_address(resolution) in sinkhole_range:
+                        is_malicious = True
+                        break
+            except dns.exception.Timeout:
+                # If primary DNS times out, try backup DNS
+                resolver.nameservers = [backup_dns]
+                try:
+                    answers = resolver.resolve(observable, "A")
+                    for rdata in answers:
+                        resolution = rdata.to_text()
+                        if ipaddress.ip_address(resolution) in sinkhole_range:
+                            is_malicious = True
+                            break
+                except dns.exception.Timeout:
+                    raise AnalyzerRunException("Connection to UltraDNS failed")
+                except Exception as e:
+                    raise Exception(f"DNS query failed for {backup_dns}: {e}")
+            except Exception as e:
+                raise Exception(f"DNS query failed for {primary_dns}: {e}")
+
+        except Exception as e:
+            raise AnalyzerRunException(f"An error occurred: {e}")
+
+        return malicious_detector_response(self.observable_name, is_malicious)
+
+    @classmethod
+    def _monkeypatch(cls):
+        patches = [
+            if_mock_connections(
+                patch("dns.resolver.Resolver.resolve", return_value=["156.154.112.16"]),
+            )
+        ]
+        return super()._monkeypatch(patches=patches)