Abuse Whois Analyzer (Closes #2308)

api_app/analyzers_manager/migrations/0147_analyzer_config_abusewhois.py

@@ -0,0 +1,125 @@
+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": "abusewhois.AbuseWHOIS",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "AbuseWhois",
+    "description": "A Sigma and RDAP/Whois-based abuse contacts finder.",
+    "disabled": False,
+    "soft_time_limit": 30,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "AMBER",
+    "observable_supported": ["ip", "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", "0146_analyzer_config_wad"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/observable_analyzers/abusewhois.py

@@ -0,0 +1,199 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+import asyncio
+
+from abuse_whois import get_abuse_contacts
+
+from api_app.analyzers_manager import classes
+
+
+class AbuseWHOIS(classes.ObservableAnalyzer):
+
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
+    def _clean_contact_info(self, contact):
+        """Remove null values and replace with REDACTED if appropriate"""
+        if not any(contact.values()):
+            return {"status": "REDACTED FOR PRIVACY"}
+        return {k: v for k, v in contact.items() if v is not None}
+
+    def _parse_raw_whois_text(self, raw_text):
+        """Extract network information from raw WHOIS text"""
+        info = {}
+        for line in raw_text.split("\n"):
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+
+            if ":" in line:
+                key, value = line.split(":", 1)
+                key = key.strip()
+                value = value.strip()
+                if value:
+                    info[key] = value
+
+        return info
+
+    def _format_ip_data(self, result):
+        """Format IP address WHOIS data"""
+        raw_info = self._parse_raw_whois_text(result.records.ip_address.raw_text)
+
+        return {
+            "network": {
+                "address": result.address,
+                "hostname": result.hostname,
+                "ip_address": result.ip_address,
+                "range": raw_info.get("NetRange"),
+                "cidr": raw_info.get("CIDR"),
+                "name": raw_info.get("NetName"),
+                "type": raw_info.get("NetType"),
+                "origin_as": raw_info.get("OriginAS"),
+            },
+            "organization": {
+                "name": raw_info.get("OrgName"),
+                "id": raw_info.get("OrgId"),
+                "address": raw_info.get("Address"),
+                "city": raw_info.get("City"),
+                "state": raw_info.get("StateProv"),
+                "postal_code": raw_info.get("PostalCode"),
+                "country": raw_info.get("Country"),
+                "registration_date": raw_info.get("RegDate"),
+                "last_updated": raw_info.get("Updated"),
+            },
+            "contacts": {
+                "abuse": {
+                    "email": raw_info.get("OrgAbuseEmail"),
+                    "phone": raw_info.get("OrgAbusePhone"),
+                    "name": raw_info.get("OrgAbuseName"),
+                },
+                "technical": {
+                    "email": raw_info.get("OrgTechEmail"),
+                    "phone": raw_info.get("OrgTechPhone"),
+                    "name": raw_info.get("OrgTechName"),
+                },
+            },
+        }
+
+    def _format_domain_data(self, result):
+        """Format domain WHOIS data"""
+        return {
+            "domain": {
+                "name": result.address,
+                "ip_address": result.ip_address,
+                "registrar": {
+                    "provider": result.registrar.provider if result.registrar else None,
+                    "email": result.registrar.address if result.registrar else None,
+                    "type": result.registrar.type if result.registrar else None,
+                },
+            },
+            "domain_info": {
+                "nameservers": (
+                    result.records.domain.name_servers if result.records.domain else []
+                ),
+                "statuses": (
+                    result.records.domain.statuses if result.records.domain else []
+                ),
+                "expires_at": (
+                    result.records.domain.expires_at.isoformat()
+                    if result.records.domain and result.records.domain.expires_at
+                    else None
+                ),
+                "updated_at": (
+                    result.records.domain.updated_at.isoformat()
+                    if result.records.domain and result.records.domain.updated_at
+                    else None
+                ),
+            },
+            "contacts": {
+                "registrant": self._clean_contact_info(
+                    {
+                        "organization": (
+                            result.records.domain.registrant.organization
+                            if result.records.domain
+                            else None
+                        ),
+                        "email": (
+                            result.records.domain.registrant.email
+                            if result.records.domain
+                            else None
+                        ),
+                        "name": (
+                            result.records.domain.registrant.name
+                            if result.records.domain
+                            else None
+                        ),
+                        "telephone": (
+                            result.records.domain.registrant.telephone
+                            if result.records.domain
+                            else None
+                        ),
+                    }
+                ),
+                "abuse": self._clean_contact_info(
+                    {
+                        "email": (
+                            result.records.domain.abuse.email
+                            if result.records.domain
+                            else None
+                        ),
+                        "telephone": (
+                            result.records.domain.abuse.telephone
+                            if result.records.domain
+                            else None
+                        ),
+                    }
+                ),
+                "technical": self._clean_contact_info(
+                    {
+                        "organization": (
+                            result.records.domain.tech.organization
+                            if result.records.domain
+                            else None
+                        ),
+                        "email": (
+                            result.records.domain.tech.email
+                            if result.records.domain
+                            else None
+                        ),
+                        "name": (
+                            result.records.domain.tech.name
+                            if result.records.domain
+                            else None
+                        ),
+                        "telephone": (
+                            result.records.domain.tech.telephone
+                            if result.records.domain
+                            else None
+                        ),
+                    }
+                ),
+            },
+        }
+
+    async def _get_whois_data(self):
+        """Get and format WHOIS data"""
+        result = await get_abuse_contacts(self.observable_name)
+
+        # Determine if this is an IP address or domain lookup and format accordingly
+        formatted_data = (
+            self._format_ip_data(result)
+            if result.records.domain is None
+            else self._format_domain_data(result)
+        )
+
+        # Remove any remaining null values at the top level
+        return {k: v for k, v in formatted_data.items() if v is not None}
+
+    def run(self):
+        """Run the analyzer"""
+        report = asyncio.run(self._get_whois_data())
+        return report
+
+    @classmethod
+    def _monkeypatch(cls):
+        patches = []
+        return super()._monkeypatch(patches=patches)

docker/Dockerfile

@@ -52,7 +52,9 @@ WORKDIR $PYTHONPATH
 
 RUN pip3 install --no-cache-dir --compile -r project-requirements.txt \
     && pip3 install --no-cache-dir pycti==${PYCTI_VERSION} \
-    && pip3 install --no-cache-dir --compile -r certego-requirements.txt
+    && pip3 install --no-cache-dir --compile -r certego-requirements.txt 
+
+RUN pip3 install abuse_whois==0.10.1 
 
 COPY . $PYTHONPATH
 

requirements/project-requirements.txt

@@ -88,7 +88,7 @@ docxpy==0.8.5
 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
-
+starlette==0.45.2
 # this is required because XLMMacroDeobfuscator does not pin the following packages
 pyxlsb2==0.0.8
 xlrd2==1.3.4