Extract IPs and CIDRs from SPF TXT records

bbot/core/helpers/dns/helpers.py

@@ -1,6 +1,12 @@
 import logging
 
-from bbot.core.helpers.regexes import dns_name_extraction_regex
+from bbot.core.helpers.regexes import (
+    dns_name_extraction_regex,
+    ip_range_regexes,
+    ipv4_regex,
+    ipv6_regex,
+    spf_ip_mechanism_regex,
+)
 from bbot.core.helpers.misc import clean_dns_record
 
 log = logging.getLogger("bbot.core.helpers.dns")
@@ -17,23 +23,48 @@ def extract_targets(record):
     already extracted the embedded names in Rust -- we just hand those back.
 
     For TXT records we additionally apply a hostname regex to the text content,
-    since SPF / DKIM / similar TXT payloads commonly embed hostnames worth
-    pivoting on. That regex extraction is BBOT-specific and stays here.
+    since SPF / DKIM / similar TXT payloads commonly embed hostnames. SPF records
+    (v=spf1) also get the IPs and CIDR ranges from their ip4:/ip6: mechanisms
+    extracted; SPF macros are skipped since they are evaluation-time templates,
+    not literal targets. That regex extraction is BBOT-specific and stays here.
     """
     results = set()
     for rdtype, host in record.extract_targets():
         cleaned = clean_dns_record(host)
         if cleaned:
             results.add((rdtype, cleaned))
 
-    # TXT: pull additional hostnames out of the free-form text content
+    # TXT: pull additional targets out of the free-form text content
     is_txt = "TXT" in record.rdata
     if is_txt:
         text = record.to_text()
-        for match in dns_name_extraction_regex.finditer(text):
-            cleaned = clean_dns_record(text[match.start() : match.end()])
-            if cleaned:
-                results.add(("TXT", cleaned))
+        if "v=spf1" in text.lower():
+            # SPF (RFC 7208): ip4:/ip6: mechanisms embed IPs and CIDR ranges
+            for token in text.split():
+                # macros (e.g. "exists:%{i}.evilcorp.com") are evaluation-time templates
+                if "%" in token:
+                    continue
+                mechanism = spf_ip_mechanism_regex.match(token)
+                if mechanism:
+                    value = token[mechanism.end() :]
+                    if (
+                        any(r.fullmatch(value) for r in ip_range_regexes)
+                        or ipv4_regex.fullmatch(value)
+                        or ipv6_regex.fullmatch(value)
+                    ):
+                        results.add(("TXT", value))
+                    continue
+                # hostnames, e.g. "include:cloudprovider.com", "redirect=_spf.example.com"
+                for match in dns_name_extraction_regex.finditer(token):
+                    cleaned = clean_dns_record(token[match.start() : match.end()])
+                    if cleaned:
+                        results.add(("TXT", cleaned))
+        else:
+            # non-SPF TXT (DKIM, verification strings, etc.): hostnames only
+            for match in dns_name_extraction_regex.finditer(text):
+                cleaned = clean_dns_record(text[match.start() : match.end()])
+                if cleaned:
+                    results.add(("TXT", cleaned))
 
     return results
 

bbot/core/helpers/regexes.py

@@ -53,6 +53,9 @@
 )
 ip_range_regexes = [re.compile(r, re.I) for r in _ip_range_regexes]
 
+# SPF qualifier + ip4:/ip6: mechanism prefix, e.g. the "ip4:" in "ip4:1.2.3.0/24" (RFC 7208)
+spf_ip_mechanism_regex = re.compile(r"^[+\-~?]?ip[46]:", re.I)
+
 # all dns names including IP addresses and bare hostnames (e.g. "localhost")
 _dns_name_regex = r"(?:\w(?:[\w-]{0,100}\w)?\.?)+(?:[xX][nN]--)?[^\W_]{1,63}\.?"
 # dns names with periods (e.g. "www.example.com")

bbot/modules/internal/dnsresolve.py

@@ -11,7 +11,7 @@
 
 class DNSResolve(BaseInterceptModule):
     watched_events = ["*"]
-    produced_events = ["DNS_NAME", "IP_ADDRESS", "RAW_DNS_RECORD"]
+    produced_events = ["DNS_NAME", "IP_ADDRESS", "IP_RANGE", "RAW_DNS_RECORD"]
     meta = {"description": "Perform DNS resolution", "created_date": "2022-04-08", "author": "@TheTechromancer"}
     _priority = 1
     scope_distance_modifier = None
@@ -203,7 +203,8 @@ async def emit_dns_children(self, event):
                 try:
                     child_event = self.scan.make_event(
                         child_host,
-                        "DNS_NAME",
+                        # auto-detect the type; SPF TXT children can be IP_ADDRESS / IP_RANGE
+                        None,
                         module=module,
                         parent=event,
                         context=f"{rdtype} record for {event.host} contains {{event.type}}: {{event.host}}",

bbot/modules/internal/speculate.py

@@ -88,8 +88,9 @@ async def handle_event(self, event):
         event_in_scope_distance = event.scope_distance <= (self.scan.scope_search_distance + 1)
         speculate_open_ports = self._should_emit_open_ports(event) and event_in_scope_distance
 
-        # generate individual IP addresses from IP range
-        if event.type == "IP_RANGE":
+        # generate individual IP addresses from IP range (only within search distance;
+        # children land at distance+1, so out-of-range CIDRs produce unreachable churn)
+        if event.type == "IP_RANGE" and event.scope_distance <= self.scan.scope_search_distance:
             net = ipaddress.ip_network(event.data)
             num_ips = net.num_addresses
             ip_range_max_hosts = self.config.get("ip_range_max_hosts", 65536)

bbot/test/test_step_1/test_dns.py

@@ -70,6 +70,74 @@ async def test_dns_engine(bbot_scanner):
     assert not extracted
 
 
+@pytest.mark.asyncio
+async def test_extract_targets_spf_ips(bbot_scanner):
+    """SPF TXT records: extract IPs and CIDRs (ip4:/ip6:) alongside hostnames."""
+    scan = bbot_scanner()
+    await scan._prep()
+    await scan.helpers.dns._mock_dns(
+        {
+            "evilcorp.com": {
+                "TXT": [
+                    '"v=spf1 ip4:1.2.3.4 ip4:5.6.7.0/24 ip6:2001:db8::/48 include:cloudprovider.com exists:%{i}.sniff.evilcorp.com -all"'
+                ],
+            },
+            "nospf.evilcorp.com": {
+                "TXT": ['"google-site-verification=abc123 contact admin.evilcorp.com ip4:9.9.9.0/24"'],
+            },
+        }
+    )
+    response = await scan.helpers.dns.resolve_full("evilcorp.com", "TXT")
+    extracted = set()
+    for ans in response.response.answers:
+        extracted.update(extract_targets(ans))
+    hosts = {host for _, host in extracted}
+    assert "1.2.3.4" in hosts, "single IPv4 from ip4: mechanism not extracted"
+    assert "5.6.7.0/24" in hosts, "IPv4 CIDR from ip4: mechanism not extracted"
+    assert "2001:db8::/48" in hosts, "IPv6 CIDR from ip6: mechanism not extracted"
+    assert {"cloudprovider.com"} <= hosts, "include: hostname not extracted"
+    assert "5.6.7.0" not in hosts, "bare network address must not be emitted alongside its CIDR"
+    assert not any("sniff" in host or "%" in host for host in hosts), "SPF macro mechanisms must be skipped"
+
+    # non-SPF TXT records are not eligible for IP/CIDR extraction, only hostnames
+    response = await scan.helpers.dns.resolve_full("nospf.evilcorp.com", "TXT")
+    extracted = set()
+    for ans in response.response.answers:
+        extracted.update(extract_targets(ans))
+    hosts = {host for _, host in extracted}
+    assert {"admin.evilcorp.com"} <= hosts, "hostname extraction from non-SPF TXT must be preserved"
+    assert "9.9.9.0/24" not in hosts, "CIDR extraction must be gated on v=spf1"
+
+
+@pytest.mark.asyncio
+async def test_extract_targets_spf_ips_end_to_end(bbot_scanner):
+    """SPF TXT record all the way through the scan: IPs/CIDRs emitted as IP_ADDRESS / IP_RANGE events."""
+    scan = bbot_scanner("evilcorp.com", config={"dns": {"minimal": False}, "scope": {"report_distance": 1}})
+    await scan._prep()
+    await scan.helpers.dns._mock_dns(
+        {
+            "evilcorp.com": {
+                "TXT": [
+                    '"v=spf1 ip4:1.2.3.4 ip4:5.6.7.0/24 ip6:2001:db8::/48 include:cloudprovider.com exists:%{i}.sniff.evilcorp.com -all"'
+                ],
+            },
+            "cloudprovider.com": {"A": ["4.3.2.1"]},
+        }
+    )
+    events = [e async for e in scan.async_start()]
+    ip_addresses = {e.data for e in events if e.type == "IP_ADDRESS"}
+    ip_ranges = {e.data for e in events if e.type == "IP_RANGE"}
+    dns_names = {e.data for e in events if e.type == "DNS_NAME"}
+    assert "1.2.3.4" in ip_addresses, "SPF ip4: address was not emitted as an IP_ADDRESS event"
+    assert "5.6.7.0/24" in ip_ranges, "SPF ip4: CIDR was not emitted as an IP_RANGE event"
+    assert "2001:db8::/48" in ip_ranges, "SPF ip6: CIDR was not emitted as an IP_RANGE event"
+    assert {"cloudprovider.com"} <= dns_names, "SPF include: hostname was not emitted as a DNS_NAME event"
+    assert not any("sniff" in str(e.data) for e in events if e.type in ("DNS_NAME", "IP_ADDRESS", "IP_RANGE")), (
+        "SPF macro mechanism must not produce DNS_NAME/IP events"
+    )
+    await scan._cleanup()
+
+
 @pytest.mark.asyncio
 async def test_dns_resolution(bbot_scanner):
     """Multi-rdtype resolution + SPF affiliate tagging end-to-end."""

bbot/test/test_step_2/module_tests/test_module_speculate.py

@@ -81,6 +81,56 @@ def check(self, module_test, events):
         )
 
 
+class TestSpeculate_OutOfScopeIPRange(ModuleTestBase):
+    """Out-of-scope IP ranges from SPF (or any source) should not be enumerated into individual IPs."""
+
+    targets = ["evilcorp.com"]
+    modules_overrides = ["speculate"]
+    config_overrides = {"dns": {"minimal": False}}
+
+    async def setup_after_prep(self, module_test):
+        await module_test.mock_dns(
+            {
+                "evilcorp.com": {
+                    "A": ["127.0.254.1"],
+                    "TXT": ['"v=spf1 ip4:192.168.0.0/24 ~all"'],
+                },
+            }
+        )
+
+        from bbot.modules.base import BaseModule
+
+        class IPCollector(BaseModule):
+            _name = "ip_collector"
+            watched_events = ["IP_ADDRESS"]
+            scope_distance_modifier = 10
+            accept_dupes = True
+
+            async def setup(self):
+                self.events = []
+                return True
+
+            async def handle_event(self, event):
+                self.events.append(event)
+
+        collector = IPCollector(module_test.scan)
+        await collector.setup()
+        module_test.scan.modules["ip_collector"] = collector
+
+    def check(self, module_test, events):
+        import ipaddress
+
+        net = ipaddress.ip_network("192.168.0.0/24")
+        speculated_ips = [
+            e
+            for e in module_test.scan.modules["ip_collector"].events
+            if e.type == "IP_ADDRESS" and ipaddress.ip_address(e.data) in net
+        ]
+        assert len(speculated_ips) == 0, (
+            f"speculate enumerated {len(speculated_ips)} IPs from out-of-scope IP_RANGE 192.168.0.0/24"
+        )
+
+
 class TestSpeculate_OpenPorts_Portscanner(TestSpeculate_OpenPorts):
     targets = ["evilcorp.com"]
     modules_overrides = ["speculate", "certspotter", "portscan"]