Mullvad_DNS (New Analyzer)

api_app/analyzers_manager/migrations/0152_analyzer_config_mullvad_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_malicious_detectors.mullvad_dns.MullvadDNSAnalyzer",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "Mullvad_DNS",
+    "description": "[Mullvad_DNS](https://github.com/mullvad/dns-blocklists) is an analyzer that queries Mullvad's DNS-over-HTTPS service (using the 'base' endpoint) to check a domain's DNS records. \r\nIt supports two modes:\r\n- 'query': returns raw DNS answer data.\r\n- 'malicious': interprets an NXDOMAIN (rcode==3) as the domain being blocked (i.e., malicious).",
+    "disabled": False,
+    "soft_time_limit": 60,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "RED",
+    "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_malicious_detectors.mullvad_dns.MullvadDNSAnalyzer",
+            "base_path": "api_app.analyzers_manager.observable_analyzers",
+        },
+        "name": "mode",
+        "type": "str",
+        "description": "'query': returns raw DNS answer data.\r\n'malicious': interprets an NXDOMAIN (rcode==3) as the domain being blocked (i.e., malicious).",
+        "is_secret": False,
+        "required": False,
+    }
+]
+
+values = [
+    {
+        "parameter": {
+            "python_module": {
+                "module": "dns.dns_malicious_detectors.mullvad_dns.MullvadDNSAnalyzer",
+                "base_path": "api_app.analyzers_manager.observable_analyzers",
+            },
+            "name": "mode",
+            "type": "str",
+            "description": "'query': returns raw DNS answer data.\r\n'malicious': interprets an NXDOMAIN (rcode==3) as the domain being blocked (i.e., malicious).",
+            "is_secret": False,
+            "required": False,
+        },
+        "analyzer_config": "Mullvad_DNS",
+        "connector_config": None,
+        "visualizer_config": None,
+        "ingestor_config": None,
+        "pivot_config": None,
+        "for_organization": False,
+        "value": "query",
+        "updated_at": "2025-02-20T16:14:25.192983Z",
+        "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", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0151_analyzer_config_ipquery"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

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

@@ -0,0 +1,127 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+import base64
+import logging
+from urllib.parse import urlparse
+
+import dns.message
+import httpx
+
+from api_app.analyzers_manager.classes import ObservableAnalyzer
+from api_app.analyzers_manager.exceptions import AnalyzerConfigurationException
+from api_app.analyzers_manager.observable_analyzers.dns.dns_responses import (
+    malicious_detector_response,
+)
+from api_app.choices import Classification
+from tests.mock_utils import MockUpResponse, if_mock_connections, patch
+
+logger = logging.getLogger(__name__)
+
+
+class MullvadDNSAnalyzer(ObservableAnalyzer):
+    """
+    MullvadDNSAnalyzer:
+
+    This analyzer queries Mullvad's DNS-over-HTTPS service (using the "base" endpoint)
+    to check a domain's DNS records. It supports two modes:
+      - "query": returns raw DNS answer data.
+      - "malicious": interprets an NXDOMAIN (rcode==3) as the domain being blocked (i.e., malicious).
+    """
+
+    url = "https://base.dns.mullvad.net/dns-query"
+    mode: str = "query"
+
+    def update(self):
+        pass
+
+    @staticmethod
+    def encode_query(observable: str) -> str:
+        """
+        Constructs a DNS query for the given observable (domain) for an A record,
+        converts it to wire format, and encodes it in URL-safe base64.
+        """
+        logger.info(f"Encoding DNS query for {observable}")
+        query = dns.message.make_query(observable, dns.rdatatype.A)
+        wire_query = query.to_wire()
+        encoded_query = (
+            base64.urlsafe_b64encode(wire_query).rstrip(b"=").decode("ascii")
+        )
+        logger.info(f"Mullvad_DNS encoded query for {observable}: {encoded_query}")
+        return encoded_query
+
+    def run(self):
+        """
+        Executes the analyzer:
+          - Validates the observable type (DOMAIN or URL).
+          - For URLs, extracts the hostname.
+          - Encodes a DNS "A" record query.
+          - Makes an HTTP GET request to the Mullvad DoH endpoint.
+          - Parses the DNS response.
+          - Depending on the configured mode ("query" or "malicious"), returns either raw data or a flagged result.
+        """
+        observable = self.observable_name
+
+        if self.observable_classification == Classification.URL:
+            logger.debug(f"Mullvad_DNS extracting hostname from URL {observable}")
+            hostname = urlparse(observable).hostname
+            observable = hostname
+
+        encoded_query = self.encode_query(observable)
+        complete_url = f"{self.url}?dns={encoded_query}"
+        logger.info(f"Requesting Mullvad DNS for {observable} at: {complete_url}")
+
+        try:
+            response = httpx.Client(http2=True).get(
+                complete_url,
+                headers={"accept": "application/dns-message"},
+                timeout=30.0,
+            )
+            response.raise_for_status()
+        except httpx.HTTPError as e:
+            logger.error(f"HTTP error: {e}")
+            raise AnalyzerConfigurationException(f"Failed to query Mullvad DNS: {e}")
+
+        dns_response = dns.message.from_wire(response.content)
+
+        if self.mode == "malicious":
+            malicious = dns_response.rcode() == 3
+            return malicious_detector_response(
+                observable=observable,
+                malicious=malicious,
+                note=f"Domain is {'' if malicious else 'not '}blocked by Mullvad DNS content filtering.",
+            )
+
+        elif self.mode == "query":
+            answers = dns_response.answer
+            data = [str(rrset) for rrset in answers] if answers else []
+            return {
+                "status": "success",
+                "data": data,
+                "message": f"DNS query for {observable} completed successfully.",
+            }
+        else:
+            raise AnalyzerConfigurationException(
+                f"Invalid mode: {self.mode}. Must be 'query' or 'malicious'."
+            )
+
+    @classmethod
+    def _monkeypatch(cls):
+        patches = [
+            if_mock_connections(
+                patch(
+                    "httpx.Client.get",
+                    return_value=MockUpResponse(
+                        {
+                            "status": "success",
+                            "data": "example.com. 236 IN A 23.215.0.138",
+                            "message": "DNS query for example.com completed successfully.",
+                        },
+                        200,
+                        content=b"pn\x01\x03\x00\x01\x00\x00\x00\x00\x00\x00\x07example\x03com\x00\x00\x01\x00\x01",
+                    ),
+                )
+            )
+        ]
+        return super()._monkeypatch(patches=patches)

requirements/project-requirements.txt

@@ -89,6 +89,10 @@ pylnk3==0.4.2
 androguard==3.4.0a1 # version >=4.x of androguard raises a dependency conflict with quark-engine==25.1.1
 wad==0.4.6
 
+# httpx required for HTTP/2 support (Mullvad DNS rejects HTTP/1.1 with protocol errors)
+httpx[http2]==0.28.1
+
+
 # this is required because XLMMacroDeobfuscator does not pin the following packages
 pyxlsb2==0.0.8
 xlrd2==1.3.4