[GSoC 2026] core(datamodel): populate the reconciled verdict for key-free domain/URL analyzers

api_app/analyzers_manager/migrations/0195_data_model_key_free_detectors.py

@@ -0,0 +1,51 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+from django.db import migrations
+
+# Single reviewable reliability table. $-prefixed keys are written as literal
+# constants (analyzers_manager/models.py:92-95). Conditionality lives in each
+# analyzer's _do_create_data_model gate, so a non-hit produces no data model.
+# Reliability tiers reflect source authority: Google Safe Browsing/WebRisk (8)
+# and Spamhaus (7) rank above the DNS-resolver blocklists (6).
+MAPPINGS = {
+    "GoogleSafebrowsing": {"$malicious": "evaluation", "$8": "reliability"},
+    "GoogleWebRisk": {"$malicious": "evaluation", "$8": "reliability"},
+    "Spamhaus_WQS": {"$malicious": "evaluation", "$7": "reliability"},
+    "AdGuard": {"$malicious": "evaluation", "$6": "reliability"},
+    "Quad9_Malicious_Detector": {"$malicious": "evaluation", "$6": "reliability"},
+    "CloudFlare_Malicious_Detector": {"$malicious": "evaluation", "$6": "reliability"},
+    "CleanBrowsing_Malicious_Detector": {"$malicious": "evaluation", "$6": "reliability"},
+    "UltraDNS_Malicious_Detector": {"$malicious": "evaluation", "$6": "reliability"},
+    "DNS4EU_Malicious_Detector": {"$malicious": "evaluation", "$6": "reliability"},
+    "Mullvad_DNS": {"$malicious": "evaluation", "$6": "reliability"},
+}
+
+
+def apply_mappings(apps, schema_editor):
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    for name, mapping in MAPPINGS.items():
+        ac = AnalyzerConfig.objects.filter(name=name).first()
+        if not ac:
+            continue
+        ac.mapping_data_model = mapping
+        ac.save()
+
+
+def revert_mappings(apps, schema_editor):
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    for name in MAPPINGS:
+        ac = AnalyzerConfig.objects.filter(name=name).first()
+        if not ac:
+            continue
+        ac.mapping_data_model = {}
+        ac.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("analyzers_manager", "0194_analyzer_config_rdap"),
+    ]
+    operations = [
+        migrations.RunPython(apply_mappings, revert_mappings),
+    ]

api_app/analyzers_manager/migrations/0196_data_model_phishing_lists.py

@@ -0,0 +1,41 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+from django.db import migrations
+
+# Same declarative pattern as 0195: $-prefixed keys are written as literal
+# constants (analyzers_manager/models.py:92-95). A listing hit is gated in each
+# analyzer's _do_create_data_model, so a miss produces no data model.
+MAPPINGS = {
+    "PhishingArmy": {"$malicious": "evaluation", "$6": "reliability"},
+    "Phishstats": {"$malicious": "evaluation", "$6": "reliability"},
+}
+
+
+def apply_mappings(apps, schema_editor):
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    for name, mapping in MAPPINGS.items():
+        ac = AnalyzerConfig.objects.filter(name=name).first()
+        if not ac:
+            continue
+        ac.mapping_data_model = mapping
+        ac.save()
+
+
+def revert_mappings(apps, schema_editor):
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    for name in MAPPINGS:
+        ac = AnalyzerConfig.objects.filter(name=name).first()
+        if not ac:
+            continue
+        ac.mapping_data_model = {}
+        ac.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("analyzers_manager", "0195_data_model_key_free_detectors"),
+    ]
+    operations = [
+        migrations.RunPython(apply_mappings, revert_mappings),
+    ]

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

@@ -8,11 +8,12 @@
 
 from ..dns_responses import malicious_detector_response
 from ..doh_mixin import DoHMixin
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 logger = logging.getLogger(__name__)
 
 
-class AdGuard(DoHMixin, classes.ObservableAnalyzer):
+class AdGuard(MaliciousDetectorResponseDataModelMixin, DoHMixin, classes.ObservableAnalyzer):
     """Check if a domain is malicious by AdGuard public resolver."""
 
     url: str = "https://dns.adguard-dns.com/dns-query"

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

@@ -13,9 +13,10 @@
 from api_app.choices import Classification
 
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 
-class CleanBrowsingMaliciousDetector(classes.ObservableAnalyzer):
+class CleanBrowsingMaliciousDetector(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     """Resolve a DNS query with CleanBrowsing security endpoint,
     Blocked domains return NXDOMAIN with SOA from cleanbrowsing.rpz.noc.org.
     """

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

@@ -12,9 +12,10 @@
 from api_app.choices import Classification
 
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 
-class CloudFlareMaliciousDetector(classes.ObservableAnalyzer):
+class CloudFlareMaliciousDetector(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     """Resolve a DNS query with CloudFlare security endpoint,
     if response is 0.0.0.0 the domain in DNS query is malicious.
     """

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

@@ -0,0 +1,17 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+class MaliciousDetectorResponseDataModelMixin:
+    """Emit a DataModel only on a real malicious hit.
+
+    These analyzers map the constant ``$malicious -> evaluation`` in their
+    ``mapping_data_model``, which writes ``evaluation = "malicious"`` unconditionally
+    whenever a data model is created. So a clean lookup (``malicious: false``), a
+    timeout, or a failure note would otherwise be stamped MALICIOUS. Gating creation
+    on ``report["malicious"] is True`` makes a non-hit produce no data model (silent);
+    it must never map a non-hit to trusted.
+    """
+
+    def _do_create_data_model(self) -> bool:
+        return super()._do_create_data_model() and self.report.report.get("malicious") is True

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

@@ -9,11 +9,12 @@
 
 from ..dns4eu_base import DNS4EUBase
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 logger = logging.getLogger(__name__)
 
 
-class DNS4EUMaliciousDetector(DNS4EUBase):
+class DNS4EUMaliciousDetector(MaliciousDetectorResponseDataModelMixin, DNS4EUBase):
     url = "https://protective.joindns4.eu/dns-query"
 
     # DNS4EU blocks by returning 0.0.0.0 or specific sinkhole IPs.

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

@@ -16,10 +16,12 @@
 )
 from api_app.choices import Classification
 
+from .data_model import MaliciousDetectorResponseDataModelMixin
+
 logger = logging.getLogger(__name__)
 
 
-class WebRisk(classes.ObservableAnalyzer):
+class WebRisk(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     """Check if observable analyzed is marked as malicious by Google WebRisk API
 
     Get these secrets from a Service Account valid file.

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

@@ -11,6 +11,7 @@
 from api_app.analyzers_manager.exceptions import AnalyzerRunException
 
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 
 class MockUpSafeBrowsing:
@@ -27,7 +28,7 @@ def lookup_urls(urls: List[str]) -> Dict:
         }
 
 
-class GoogleSF(classes.ObservableAnalyzer):
+class GoogleSF(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     """Check if observable analyzed is marked as malicious for Google SafeBrowsing"""
 
     _api_key_name: str

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

@@ -14,10 +14,12 @@
 )
 from api_app.analyzers_manager.observable_analyzers.dns.doh_mixin import DoHMixin
 
+from .data_model import MaliciousDetectorResponseDataModelMixin
+
 logger = logging.getLogger(__name__)
 
 
-class MullvadDNSAnalyzer(DoHMixin, ObservableAnalyzer):
+class MullvadDNSAnalyzer(MaliciousDetectorResponseDataModelMixin, DoHMixin, ObservableAnalyzer):
     """
     MullvadDNSAnalyzer:
 

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

@@ -11,11 +11,12 @@
 
 from ..dns_responses import malicious_detector_response
 from ..quad9_base import Quad9Base
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 logger = logging.getLogger(__name__)
 
 
-class Quad9MaliciousDetector(Quad9Base, classes.ObservableAnalyzer):
+class Quad9MaliciousDetector(MaliciousDetectorResponseDataModelMixin, Quad9Base, classes.ObservableAnalyzer):
     """Check if a domain is malicious by Quad9 public resolver.
     Quad9 does not answer in the case a malicious domain is queried.
     However, we need to perform another check to understand if that domain was blocked

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

@@ -7,11 +7,12 @@
 from api_app.choices import Classification
 
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 logger = logging.getLogger(__name__)
 
 
-class SpamhausWQS(classes.ObservableAnalyzer):
+class SpamhausWQS(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     url: str = "https://apibl.spamhaus.net/lookup/v1"
     _api_key: str = None
 

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

@@ -8,9 +8,10 @@
 from api_app.choices import Classification
 
 from ..dns_responses import malicious_detector_response
+from .data_model import MaliciousDetectorResponseDataModelMixin
 
 
-class UltraDNSMaliciousDetector(classes.ObservableAnalyzer):
+class UltraDNSMaliciousDetector(MaliciousDetectorResponseDataModelMixin, classes.ObservableAnalyzer):
     """Resolve a DNS query with UltraDNS servers,
     if the response falls within the sinkhole range, the domain is malicious.
     """

api_app/analyzers_manager/observable_analyzers/phishing_army.py

@@ -29,6 +29,11 @@ def run(self):
         found = PhishingArmyDomain.objects.filter(domain=to_analyze_observable).exists()
         return {"found": found, "link": self.url}
 
+    # Gate on a real listing hit: the $malicious mapping constant would otherwise
+    # stamp MALICIOUS on every clean lookup.
+    def _do_create_data_model(self) -> bool:
+        return super()._do_create_data_model() and self.report.report.get("found") is True
+
     @classmethod
     def update(cls) -> bool:
         try:

api_app/analyzers_manager/observable_analyzers/phishstats.py

@@ -22,6 +22,11 @@ class PhishStats(ObservableAnalyzer):
     def update(cls) -> bool:
         pass
 
+    # Gate on a real listing hit: the $malicious mapping constant would otherwise
+    # stamp MALICIOUS on every clean lookup.
+    def _do_create_data_model(self) -> bool:
+        return super()._do_create_data_model() and bool(self.report.report.get("results"))
+
     def __build_phishstats_url(self) -> str:
         to_analyze_observable_classification = self.observable_classification
         to_analyze_observable_name = self.observable_name

api_app/analyzers_manager/observable_analyzers/phishtank.py

@@ -10,9 +10,16 @@
 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.data_model_manager.enums import DataModelEvaluations
 
 logger = logging.getLogger(__name__)
 
+# A community-verified phishing hit is a strong malicious signal; an unverified
+# listing is deliberately given reliability 5 — below MALICIOUS_RELIABILITY_FLOOR
+# (6) — so it buckets to "suspicious" rather than "malicious".
+_RELIABILITY_VERIFIED = 8
+_RELIABILITY_UNVERIFIED = 5
+
 
 class Phishtank(ObservableAnalyzer):
     _api_key_name: str
@@ -41,3 +48,17 @@ def run(self):
         except requests.RequestException as e:
             raise AnalyzerRunException(e)
         return result
+
+    # Gate: the $-mapping would stamp MALICIOUS on any created model, so only a
+    # real listing (in_database) may create one.
+    def _do_create_data_model(self) -> bool:
+        results = self.report.report.get("results") or {}
+        return super()._do_create_data_model() and results.get("in_database") is True
+
+    def _update_data_model(self, data_model) -> None:
+        super()._update_data_model(data_model)
+        results = self.report.report.get("results") or {}
+        data_model.evaluation = DataModelEvaluations.MALICIOUS.value
+        data_model.reliability = (
+            _RELIABILITY_VERIFIED if results.get("verified") is True else _RELIABILITY_UNVERIFIED
+        )

api_app/analyzers_manager/observable_analyzers/tranco.py

@@ -7,6 +7,23 @@
 
 from api_app.analyzers_manager import classes
 from api_app.choices import Classification
+from api_app.data_model_manager.enums import DataModelEvaluations
+
+# A top-1000 rank is treated as a reliable allowlist rather than as weak popularity
+# evidence: reaching the "trusted" bucket (>=8) means such a domain outranks a
+# malicious detector instead of reconciling to malicious. That is deliberate — in daily
+# incident response, false positives are the expensive failure mode, because the
+# analyst time they burn costs more than the rare true positive they hide.
+# Below the top 1000 popularity really is weak evidence, so reliability stays capped at
+# 4, strictly under the malicious detectors (6-8), and a flagged domain still
+# reconciles to malicious.
+_RANK_HIGHLY_TRUSTED = 1_000
+_RANK_TOP = 10_000
+_RANK_POPULAR = 100_000
+_RELIABILITY_HIGHLY_TRUSTED = 9
+_RELIABILITY_TOP = 4
+_RELIABILITY_POPULAR = 3
+_RELIABILITY_RANKED = 2
 
 
 class Tranco(classes.ObservableAnalyzer):
@@ -26,3 +43,20 @@ def run(self):
         response.raise_for_status()
 
         return response.json()
+
+    def _do_create_data_model(self) -> bool:
+        rank = self.report.report.get("rank")
+        return super()._do_create_data_model() and isinstance(rank, int) and rank > 0
+
+    def _update_data_model(self, data_model) -> None:
+        super()._update_data_model(data_model)
+        rank = self.report.report.get("rank")
+        data_model.evaluation = DataModelEvaluations.TRUSTED.value
+        if rank <= _RANK_HIGHLY_TRUSTED:
+            data_model.reliability = _RELIABILITY_HIGHLY_TRUSTED
+        elif rank <= _RANK_TOP:
+            data_model.reliability = _RELIABILITY_TOP
+        elif rank <= _RANK_POPULAR:
+            data_model.reliability = _RELIABILITY_POPULAR
+        else:
+            data_model.reliability = _RELIABILITY_RANKED

api_app/data_model_manager/classify.py

@@ -0,0 +1,30 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+from api_app.data_model_manager.enums import DataModelEvaluations, DataModelVerdictBuckets
+
+# Bucket boundaries (verbatim from the pre-existing visualizer logic this
+# function replaced).
+TRUSTED_RELIABILITY_FLOOR = 8
+MALICIOUS_RELIABILITY_FLOOR = 6
+
+
+def classify(evaluation: str | None, reliability: int) -> str:
+    """Map a (evaluation, reliability) pair to one of the five presentation buckets.
+
+    Single source of truth for the bucketing: the DataModel visualizer calls this
+    and so does the chatbot, so the badge and the chat always agree.
+    """
+    if evaluation == DataModelEvaluations.TRUSTED.value:
+        return (
+            DataModelVerdictBuckets.TRUSTED.value
+            if reliability >= TRUSTED_RELIABILITY_FLOOR
+            else DataModelVerdictBuckets.CLEAN.value
+        )
+    if evaluation == DataModelEvaluations.MALICIOUS.value:
+        return (
+            DataModelVerdictBuckets.MALICIOUS.value
+            if reliability >= MALICIOUS_RELIABILITY_FLOOR
+            else DataModelVerdictBuckets.SUSPICIOUS.value
+        )
+    return DataModelVerdictBuckets.NO_EVALUATION.value