diff --git a/api_app/analyzers_manager/migrations/0195_data_model_key_free_detectors.py b/api_app/analyzers_manager/migrations/0195_data_model_key_free_detectors.py new file mode 100644 index 0000000000..c220d42191 --- /dev/null +++ b/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), + ] diff --git a/api_app/analyzers_manager/migrations/0196_data_model_phishing_lists.py b/api_app/analyzers_manager/migrations/0196_data_model_phishing_lists.py new file mode 100644 index 0000000000..d00c28e047 --- /dev/null +++ b/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), + ] diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/adguard.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/adguard.py index a3ed4f0a5f..15385b6025 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/adguard.py +++ b/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" diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cleanbrowsing_malicious_detector.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cleanbrowsing_malicious_detector.py index 886d2e9ec1..db2caa738e 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cleanbrowsing_malicious_detector.py +++ b/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. """ diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cloudflare_malicious_detector.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cloudflare_malicious_detector.py index ca4fa1997a..67d7227e4a 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/cloudflare_malicious_detector.py +++ b/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. """ diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/data_model.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/data_model.py new file mode 100644 index 0000000000..f83ecc7067 --- /dev/null +++ b/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 diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/dns4eu_malicious_detector.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/dns4eu_malicious_detector.py index 3a0d66a090..f49e08a8aa 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/dns4eu_malicious_detector.py +++ b/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. diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/google_webrisk.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/google_webrisk.py index 966bcb9fe1..8dc830d9ca 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/google_webrisk.py +++ b/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. diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/googlesf.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/googlesf.py index 50bb3b6ae8..7206ccf701 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/googlesf.py +++ b/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 diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/mullvad_dns.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/mullvad_dns.py index ec5b5508b1..4ff74cc0e7 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/mullvad_dns.py +++ b/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: diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/quad9_malicious_detector.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/quad9_malicious_detector.py index 2dae328059..66164691a0 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/quad9_malicious_detector.py +++ b/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 diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/spamhaus_wqs.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/spamhaus_wqs.py index 5202da3ab0..3362026885 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/spamhaus_wqs.py +++ b/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 diff --git a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/ultradns_malicious_detector.py b/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/ultradns_malicious_detector.py index fb913ade68..32fbce4bf1 100644 --- a/api_app/analyzers_manager/observable_analyzers/dns/dns_malicious_detectors/ultradns_malicious_detector.py +++ b/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. """ diff --git a/api_app/analyzers_manager/observable_analyzers/phishing_army.py b/api_app/analyzers_manager/observable_analyzers/phishing_army.py index bb7c492a8d..08e93c8cbb 100644 --- a/api_app/analyzers_manager/observable_analyzers/phishing_army.py +++ b/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: diff --git a/api_app/analyzers_manager/observable_analyzers/phishstats.py b/api_app/analyzers_manager/observable_analyzers/phishstats.py index 9b9f949cb5..3cbb9096ad 100644 --- a/api_app/analyzers_manager/observable_analyzers/phishstats.py +++ b/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 diff --git a/api_app/analyzers_manager/observable_analyzers/phishtank.py b/api_app/analyzers_manager/observable_analyzers/phishtank.py index d942065cab..db4fa4e132 100644 --- a/api_app/analyzers_manager/observable_analyzers/phishtank.py +++ b/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 + ) diff --git a/api_app/analyzers_manager/observable_analyzers/tranco.py b/api_app/analyzers_manager/observable_analyzers/tranco.py index 00f8dbe0a5..57b8e8a936 100644 --- a/api_app/analyzers_manager/observable_analyzers/tranco.py +++ b/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 diff --git a/api_app/data_model_manager/classify.py b/api_app/data_model_manager/classify.py new file mode 100644 index 0000000000..f455560f42 --- /dev/null +++ b/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 diff --git a/api_app/data_model_manager/enums.py b/api_app/data_model_manager/enums.py index 4ac1393967..81deef6abf 100644 --- a/api_app/data_model_manager/enums.py +++ b/api_app/data_model_manager/enums.py @@ -23,6 +23,24 @@ class DataModelEvaluations(Choices): MALICIOUS = "malicious" +class DataModelVerdictBuckets(Choices): + """Presentation buckets shared by the DataModel visualizer and the chatbot. + + Only two of the five are evaluations: TRUSTED and MALICIOUS take their values from + DataModelEvaluations so the two vocabularies cannot drift apart. The other three have no + evaluation counterpart and are deliberately not added to DataModelEvaluations — CLEAN and + SUSPICIOUS mean "that evaluation, but below its reliability floor", and NO_EVALUATION means + no analyzer expressed an opinion at all. They describe how a verdict is *shown*, not what an + analyzer *concluded*. + """ + + TRUSTED = DataModelEvaluations.TRUSTED.value + CLEAN = "clean" + MALICIOUS = DataModelEvaluations.MALICIOUS.value + SUSPICIOUS = "suspicious" + NO_EVALUATION = "no evaluation" + + class DataModelKillChainPhases(Choices): RECONNAISSANCE = "reconnaissance" WEAPONIZATION = "weaponization" diff --git a/api_app/visualizers_manager/visualizers/data_model.py b/api_app/visualizers_manager/visualizers/data_model.py index e3a6ae7fc7..b84059ae7f 100644 --- a/api_app/visualizers_manager/visualizers/data_model.py +++ b/api_app/visualizers_manager/visualizers/data_model.py @@ -1,7 +1,8 @@ from logging import getLogger from typing import Dict, List -from api_app.data_model_manager.enums import DataModelEvaluations +from api_app.data_model_manager.classify import classify +from api_app.data_model_manager.enums import DataModelEvaluations, DataModelVerdictBuckets from api_app.data_model_manager.models import ( DomainDataModel, FileDataModel, @@ -296,21 +297,14 @@ def run(self) -> List[Dict]: printable_analyzer_name = data_model.analyzers_report.all().first().config.name.replace("_", " ") logger.debug(f"{printable_analyzer_name}, {data_model}") - evaluation = data_model.evaluation or "" - reliability = data_model.reliability - - if evaluation == DataModelEvaluations.TRUSTED.value: - if reliability >= 8: - trusted_data_models.append(data_model) - else: - clean_data_models.append(data_model) - elif evaluation == DataModelEvaluations.MALICIOUS.value: - if reliability >= 6: - malicious_data_models.append(data_model) - else: - suspicious_data_models.append(data_model) - else: - noeval_data_models.append(data_model) + bucket = classify(data_model.evaluation, data_model.reliability) + { + DataModelVerdictBuckets.TRUSTED.value: trusted_data_models, + DataModelVerdictBuckets.CLEAN.value: clean_data_models, + DataModelVerdictBuckets.MALICIOUS.value: malicious_data_models, + DataModelVerdictBuckets.SUSPICIOUS.value: suspicious_data_models, + DataModelVerdictBuckets.NO_EVALUATION.value: noeval_data_models, + }[bucket].append(data_model) evals_vlists = [] for evaluation, color, icon, eval_data_models in [ diff --git a/tests/api_app/analyzers_manager/test_data_model_detectors.py b/tests/api_app/analyzers_manager/test_data_model_detectors.py new file mode 100644 index 0000000000..9eddc4f62b --- /dev/null +++ b/tests/api_app/analyzers_manager/test_data_model_detectors.py @@ -0,0 +1,81 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from types import SimpleNamespace + +from kombu import uuid + +from api_app.analyzables_manager.models import Analyzable +from api_app.analyzers_manager.models import AnalyzerConfig, AnalyzerReport +from api_app.analyzers_manager.observable_analyzers.dns.dns_malicious_detectors.adguard import ( + AdGuard, +) +from api_app.choices import Classification +from api_app.models import Job +from tests import CustomTestCase + + +def _gate_stub(analyzer_cls, report_dict, mapping=None): + """Build a minimally-initialised analyzer to exercise _do_create_data_model + without a full plugin construction (mirrors how the base gate reads state).""" + analyzer = analyzer_cls.__new__(analyzer_cls) + analyzer.report = SimpleNamespace( + report=report_dict, + job=SimpleNamespace(analyzable=SimpleNamespace(classification=Classification.DOMAIN.value)), + ) + analyzer._config = SimpleNamespace(mapping_data_model=mapping or {"$malicious": "evaluation"}) + return analyzer + + +class MaliciousDetectorGateTestCase(CustomTestCase): + def test_gate_emits_only_on_real_hit(self): + self.assertTrue(_gate_stub(AdGuard, {"malicious": True})._do_create_data_model()) + + def test_gate_suppresses_clean_lookup(self): + # F1: the $malicious constant would stamp MALICIOUS on every clean lookup; + # the gate must return False so NO data model is created. + self.assertFalse(_gate_stub(AdGuard, {"malicious": False})._do_create_data_model()) + + def test_gate_suppresses_timeout_and_note(self): + self.assertFalse(_gate_stub(AdGuard, {"malicious": False, "timeout": True})._do_create_data_model()) + self.assertFalse( + _gate_stub( + AdGuard, {"malicious": False, "note": "No response from AdGuard DNS API"} + )._do_create_data_model() + ) + + +class MaliciousDetectorMappingTestCase(CustomTestCase): + def _run_mapping(self, config_name, report_dict): + an = Analyzable.objects.create(name="test.com", classification=Classification.DOMAIN) + job = Job.objects.create(analyzable=an, status=Job.STATUSES.ANALYZERS_RUNNING.value) + config = AnalyzerConfig.objects.get(name=config_name) + ar = AnalyzerReport.objects.create( + report=report_dict, + job=job, + config=config, + status=AnalyzerReport.STATUSES.SUCCESS.value, + task_id=str(uuid()), + parameters={}, + ) + job.analyzers_to_execute.set([config]) + data_model = ar.create_data_model() # report-level: applies mapping only + if data_model is not None: + data_model.refresh_from_db() + return data_model + + def test_adguard_mapping_sets_malicious_reliability_six(self): + dm = self._run_mapping("AdGuard", {"observable": "test.com", "malicious": True}) + self.assertIsNotNone(dm) + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 6) + + def test_googlewebrisk_mapping_sets_reliability_eight(self): + dm = self._run_mapping("GoogleWebRisk", {"observable": "test.com", "malicious": True}) + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 8) + + def test_spamhaus_mapping_sets_reliability_seven(self): + dm = self._run_mapping("Spamhaus_WQS", {"observable": "test.com", "malicious": True}) + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 7) diff --git a/tests/api_app/analyzers_manager/test_data_model_phishing_lists.py b/tests/api_app/analyzers_manager/test_data_model_phishing_lists.py new file mode 100644 index 0000000000..cfdb2d922d --- /dev/null +++ b/tests/api_app/analyzers_manager/test_data_model_phishing_lists.py @@ -0,0 +1,52 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from types import SimpleNamespace + +from kombu import uuid + +from api_app.analyzables_manager.models import Analyzable +from api_app.analyzers_manager.models import AnalyzerConfig, AnalyzerReport +from api_app.analyzers_manager.observable_analyzers.phishing_army import PhishingArmy +from api_app.analyzers_manager.observable_analyzers.phishstats import PhishStats +from api_app.choices import Classification +from api_app.models import Job +from tests import CustomTestCase + + +class PhishingListsGateTestCase(CustomTestCase): + @staticmethod + def _stub(analyzer_cls, report_dict): + analyzer = analyzer_cls.__new__(analyzer_cls) + analyzer.report = SimpleNamespace( + report=report_dict, + job=SimpleNamespace(analyzable=SimpleNamespace(classification=Classification.DOMAIN.value)), + ) + analyzer._config = SimpleNamespace(mapping_data_model={"$malicious": "evaluation"}) + return analyzer + + def test_phishing_army_gate(self): + self.assertTrue(self._stub(PhishingArmy, {"found": True})._do_create_data_model()) + self.assertFalse(self._stub(PhishingArmy, {"found": False})._do_create_data_model()) + + def test_phishstats_gate(self): + self.assertTrue(self._stub(PhishStats, {"results": [{"id": 1}]})._do_create_data_model()) + self.assertFalse(self._stub(PhishStats, {"results": []})._do_create_data_model()) + + def test_phishing_army_mapping_reliability_six(self): + an = Analyzable.objects.create(name="bad.com", classification=Classification.DOMAIN) + job = Job.objects.create(analyzable=an, status=Job.STATUSES.ANALYZERS_RUNNING.value) + config = AnalyzerConfig.objects.get(name="PhishingArmy") + ar = AnalyzerReport.objects.create( + report={"found": True, "link": "x"}, + job=job, + config=config, + status=AnalyzerReport.STATUSES.SUCCESS.value, + task_id=str(uuid()), + parameters={}, + ) + job.analyzers_to_execute.set([config]) + dm = ar.create_data_model() + dm.refresh_from_db() + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 6) diff --git a/tests/api_app/analyzers_manager/test_data_model_phishtank.py b/tests/api_app/analyzers_manager/test_data_model_phishtank.py new file mode 100644 index 0000000000..60cf7a4a43 --- /dev/null +++ b/tests/api_app/analyzers_manager/test_data_model_phishtank.py @@ -0,0 +1,39 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from types import SimpleNamespace + +from api_app.analyzers_manager.observable_analyzers.phishtank import Phishtank +from api_app.choices import Classification +from api_app.data_model_manager.models import DomainDataModel +from tests import CustomTestCase + + +class PhishtankDataModelTestCase(CustomTestCase): + @staticmethod + def _phishtank(results): + analyzer = Phishtank.__new__(Phishtank) + analyzer.report = SimpleNamespace( + report={"results": results}, + job=SimpleNamespace(analyzable=SimpleNamespace(classification=Classification.DOMAIN.value)), + ) + analyzer._config = SimpleNamespace(mapping_data_model={}) + return analyzer + + def test_gate_requires_in_database(self): + self.assertTrue(self._phishtank({"in_database": True})._do_create_data_model()) + self.assertFalse(self._phishtank({"in_database": False})._do_create_data_model()) + self.assertFalse(self._phishtank({})._do_create_data_model()) + + def test_verified_hit_is_reliability_eight(self): + dm = DomainDataModel() + self._phishtank({"in_database": True, "verified": True})._update_data_model(dm) + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 8) + + def test_unverified_hit_is_reliability_five(self): + dm = DomainDataModel() + dm.reliability = 0 # sentinel: prove the code sets 5, not the field default (also 5) + self._phishtank({"in_database": True, "verified": False})._update_data_model(dm) + self.assertEqual(dm.evaluation, "malicious") + self.assertEqual(dm.reliability, 5) diff --git a/tests/api_app/analyzers_manager/test_data_model_targets.py b/tests/api_app/analyzers_manager/test_data_model_targets.py new file mode 100644 index 0000000000..13ba73db32 --- /dev/null +++ b/tests/api_app/analyzers_manager/test_data_model_targets.py @@ -0,0 +1,37 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from api_app.analyzers_manager.models import AnalyzerConfig +from tests import CustomTestCase + + +class DataModelTargetsAntiRotTestCase(CustomTestCase): + MAPPED = { + "GoogleSafebrowsing": 8, + "GoogleWebRisk": 8, + "Spamhaus_WQS": 7, + "AdGuard": 6, + "Quad9_Malicious_Detector": 6, + "CloudFlare_Malicious_Detector": 6, + "CleanBrowsing_Malicious_Detector": 6, + "UltraDNS_Malicious_Detector": 6, + "DNS4EU_Malicious_Detector": 6, + "Mullvad_DNS": 6, + "PhishingArmy": 6, + "Phishstats": 6, + } + HOOKED = ["Phishtank", "Tranco"] + + def test_mapped_configs_exist_with_expected_reliability(self): + for name, reliability in self.MAPPED.items(): + config = AnalyzerConfig.objects.filter(name=name).first() + self.assertIsNotNone(config, f"missing analyzer config: {name}") + self.assertEqual(config.mapping_data_model.get("$malicious"), "evaluation", name) + self.assertEqual(config.mapping_data_model.get(f"${reliability}"), "reliability", name) + + def test_hooked_configs_exist(self): + for name in self.HOOKED: + self.assertTrue( + AnalyzerConfig.objects.filter(name=name).exists(), + f"missing analyzer config: {name}", + ) diff --git a/tests/api_app/analyzers_manager/test_data_model_tranco.py b/tests/api_app/analyzers_manager/test_data_model_tranco.py new file mode 100644 index 0000000000..4c56214db2 --- /dev/null +++ b/tests/api_app/analyzers_manager/test_data_model_tranco.py @@ -0,0 +1,47 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from types import SimpleNamespace + +from api_app.analyzers_manager.observable_analyzers.tranco import Tranco +from api_app.choices import Classification +from api_app.data_model_manager.models import DomainDataModel +from tests import CustomTestCase + + +class TrancoDataModelTestCase(CustomTestCase): + @staticmethod + def _tranco(report_dict): + analyzer = Tranco.__new__(Tranco) + analyzer.report = SimpleNamespace( + report=report_dict, + job=SimpleNamespace(analyzable=SimpleNamespace(classification=Classification.DOMAIN.value)), + ) + analyzer._config = SimpleNamespace(mapping_data_model={}) + return analyzer + + def test_gate_requires_positive_rank(self): + self.assertTrue(self._tranco({"rank": 5000})._do_create_data_model()) + self.assertFalse(self._tranco({"rank": None})._do_create_data_model()) + self.assertFalse(self._tranco({})._do_create_data_model()) + + def test_rank_bands(self): + # The top-1000 band is the only one that reaches the "trusted" bucket (floor 8); every + # boundary is pinned on both sides so a band edit cannot silently widen or shrink it. + for rank, expected in [ + (1, 9), + (1000, 9), + (1001, 4), + (10000, 4), + (10001, 3), + (100000, 3), + (100001, 2), + ]: + dm = DomainDataModel() + self._tranco({"rank": rank})._update_data_model(dm) + self.assertEqual(dm.evaluation, "trusted") + self.assertEqual(dm.reliability, expected, f"rank={rank}") + + def test_malformed_report_never_raises(self): + # a blocklist miss must never yield trusted; malformed input must not crash + self.assertFalse(self._tranco({"unexpected": "shape"})._do_create_data_model()) diff --git a/tests/api_app/data_model_manager/test_classify.py b/tests/api_app/data_model_manager/test_classify.py new file mode 100644 index 0000000000..4cb0a493ca --- /dev/null +++ b/tests/api_app/data_model_manager/test_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 django.test import SimpleTestCase + +from api_app.data_model_manager.classify import classify +from api_app.data_model_manager.enums import DataModelEvaluations + + +class ClassifyTestCase(SimpleTestCase): + def test_trusted_high_reliability_is_trusted(self): + self.assertEqual(classify(DataModelEvaluations.TRUSTED.value, 8), "trusted") + self.assertEqual(classify(DataModelEvaluations.TRUSTED.value, 10), "trusted") + + def test_trusted_below_eight_is_clean(self): + self.assertEqual(classify(DataModelEvaluations.TRUSTED.value, 7), "clean") + self.assertEqual(classify(DataModelEvaluations.TRUSTED.value, 0), "clean") + + def test_malicious_high_reliability_is_malicious(self): + self.assertEqual(classify(DataModelEvaluations.MALICIOUS.value, 6), "malicious") + self.assertEqual(classify(DataModelEvaluations.MALICIOUS.value, 10), "malicious") + + def test_malicious_below_six_is_suspicious(self): + self.assertEqual(classify(DataModelEvaluations.MALICIOUS.value, 5), "suspicious") + self.assertEqual(classify(DataModelEvaluations.MALICIOUS.value, 0), "suspicious") + + def test_none_or_unknown_is_no_evaluation(self): + self.assertEqual(classify(None, 9), "no evaluation") + self.assertEqual(classify("", 9), "no evaluation") + self.assertEqual(classify("whatever", 9), "no evaluation")