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/test/test_step_1/test_dns.py

@@ -70,6 +70,74 @@
     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" in 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" in 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" in 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."""