Merge pull request #1037 from blacklanternsecurity/baddns_module

Add Baddns Module

bbot/core/flags.py

@@ -2,6 +2,7 @@
     "active": "Makes active connections to target systems",
     "affiliates": "Discovers affiliated hostnames/domains",
     "aggressive": "Generates a large amount of network traffic",
+    "baddns": "Runs all modules from the DNS auditing tool BadDNS",
     "cloud-enum": "Enumerates cloud resources",
     "deadly": "Highly aggressive",
     "email-enum": "Enumerates email addresses",

bbot/core/helpers/dns.py

@@ -1025,28 +1025,3 @@ def _get_dummy_module(self, name):
             dummy_module.suppress_dupes = False
             self._dummy_modules[name] = dummy_module
         return dummy_module
-
-    def mock_dns(self, dns_dict):
-        if self._orig_resolve_raw is None:
-            self._orig_resolve_raw = self.resolve_raw
-
-        async def mock_resolve_raw(query, **kwargs):
-            results = []
-            errors = []
-            types = self._parse_rdtype(kwargs.get("type", ["A", "AAAA"]))
-            for t in types:
-                with suppress(KeyError):
-                    results += self._mock_table[(query, t)]
-            return results, errors
-
-        for (query, rdtype), answers in dns_dict.items():
-            if isinstance(answers, str):
-                answers = [answers]
-            for answer in answers:
-                rdata = dns.rdata.from_text("IN", rdtype, answer)
-                try:
-                    self._mock_table[(query, rdtype)].append((rdtype, rdata))
-                except KeyError:
-                    self._mock_table[(query, rdtype)] = [(rdtype, [rdata])]
-
-        self.resolve_raw = mock_resolve_raw

bbot/core/helpers/names_generator.py

@@ -93,6 +93,7 @@
     "giddy",
     "glowering",
     "glutinous",
+    "golden",
     "gothic",
     "grievous",
     "gummy",
@@ -431,6 +432,7 @@
     "gollum",
     "grace",
     "gregory",
+    "gus",
     "hagrid",
     "hannah",
     "harold",

bbot/modules/baddns.py

@@ -0,0 +1,86 @@
+from baddns.base import get_all_modules
+from baddns.lib.loader import load_signatures
+from .base import BaseModule
+
+import asyncio
+import logging
+from bbot.core.logger.logger import include_logger
+
+include_logger(logging.getLogger("baddns"))
+
+
+class baddns(BaseModule):
+    watched_events = ["DNS_NAME", "DNS_NAME_UNRESOLVED"]
+    produced_events = ["FINDING", "VULNERABILITY"]
+    flags = ["active", "safe", "web-basic", "baddns", "cloud-enum", "subdomain-hijack"]
+    meta = {"description": "Check hosts for domain/subdomain takeovers"}
+    options = {"custom_nameservers": [], "only_high_confidence": False}
+    options_desc = {
+        "custom_nameservers": "Force BadDNS to use a list of custom nameservers",
+        "only_high_confidence": "Do not emit low-confidence or generic detections",
+    }
+    max_event_handlers = 8
+    deps_pip = ["baddns~=1.1.0"]
+
+    def select_modules(self):
+        selected_modules = []
+        for m in get_all_modules():
+            if m.name in ["CNAME", "NS", "MX", "references", "TXT"]:
+                selected_modules.append(m)
+        return selected_modules
+
+    async def setup(self):
+        self.custom_nameservers = self.config.get("custom_nameservers", []) or None
+        if self.custom_nameservers:
+            self.custom_nameservers = self.helpers.chain_lists(self.custom_nameservers)
+        self.only_high_confidence = self.config.get("only_high_confidence", False)
+        self.signatures = load_signatures()
+        return True
+
+    async def handle_event(self, event):
+
+        tasks = []
+        for ModuleClass in self.select_modules():
+            module_instance = ModuleClass(
+                event.data,
+                http_client_class=self.scan.helpers.web.AsyncClient,
+                dns_client=self.scan.helpers.dns.resolver,
+                custom_nameservers=self.custom_nameservers,
+                signatures=self.signatures,
+            )
+            tasks.append((module_instance, asyncio.create_task(module_instance.dispatch())))
+
+        for module_instance, task in tasks:
+            if await task:
+                results = module_instance.analyze()
+                if results and len(results) > 0:
+                    for r in results:
+                        r_dict = r.to_dict()
+
+                        if r_dict["confidence"] in ["CONFIRMED", "PROBABLE"]:
+                            data = {
+                                "severity": "MEDIUM",
+                                "description": f"{r_dict['description']}. Confidence: [{r_dict['confidence']}] Signature: [{r_dict['signature']}] Indicator: [{r_dict['indicator']}] Trigger: [{r_dict['trigger']}] baddns Module: [{r_dict['module']}]",
+                                "host": str(event.host),
+                            }
+                            await self.emit_event(
+                                data, "VULNERABILITY", event, tags=[f"baddns-{module_instance.name.lower()}"]
+                            )
+
+                        elif r_dict["confidence"] in ["UNLIKELY", "POSSIBLE"] and not self.only_high_confidence:
+                            data = {
+                                "description": f"{r_dict['description']} Confidence: [{r_dict['confidence']}] Signature: [{r_dict['signature']}] Indicator: [{r_dict['indicator']}] Trigger: [{r_dict['trigger']}] baddns Module: [{r_dict['module']}]",
+                                "host": str(event.host),
+                            }
+                            await self.emit_event(
+                                data, "FINDING", event, tags=[f"baddns-{module_instance.name.lower()}"]
+                            )
+                        else:
+                            self.warning(f"Got unrecognized confidence level: {r['confidence']}")
+
+                        found_domains = r_dict.get("found_domains", None)
+                        if found_domains:
+                            for found_domain in found_domains:
+                                await self.emit_event(
+                                    found_domain, "DNS_NAME", event, tags=[f"baddns-{module_instance.name.lower()}"]
+                                )

bbot/modules/baddns_zone.py

@@ -0,0 +1,34 @@
+from baddns.base import get_all_modules
+from .baddns import baddns as baddns_module
+
+import logging
+from bbot.core.logger.logger import include_logger
+
+include_logger(logging.getLogger("baddns_zone"))
+
+
+class baddns_zone(baddns_module):
+    watched_events = ["DNS_NAME"]
+    produced_events = ["FINDING", "VULNERABILITY"]
+    flags = ["active", "safe", "subdomain-enum", "baddns", "cloud-enum"]
+    meta = {"description": "Check hosts for DNS zone transfers and NSEC walks"}
+    options = {"custom_nameservers": [], "only_high_confidence": False}
+    options_desc = {
+        "custom_nameservers": "Force BadDNS to use a list of custom nameservers",
+        "only_high_confidence": "Do not emit low-confidence or generic detections",
+    }
+    max_event_handlers = 8
+    deps_pip = ["baddns~=1.1.0"]
+
+    def select_modules(self):
+        selected_modules = []
+        for m in get_all_modules():
+            if m.name in ["NSEC", "zonetransfer"]:
+                selected_modules.append(m)
+        return selected_modules
+
+    # minimize nsec records feeding back into themselves
+    async def filter_event(self, event):
+        if "baddns-nsec" in event.tags or "baddns-nsec" in event.source.tags:
+            return False
+        return True