src/usfx/modules/dns_bruteforce.py (6140B)
1 """
2 DNS Brute Force Module
3
4 Performs parallel DNS resolution against a wordlist to discover subdomains.
5 Uses the internal DNS resolver for custom DNS server support.
6 """
7
8 import logging
9 from concurrent.futures import ThreadPoolExecutor, as_completed
10 from typing import Callable, Dict, Optional, Set, TYPE_CHECKING
11
12 from .base import BaseModule
13 from ..utils.wordlist import WordlistManager
14 from ..utils.wildcard import WildcardDetector
15
16 if TYPE_CHECKING:
17 from ..utils.dns_resolver import InternalDNSResolver
18
19 logger = logging.getLogger(__name__)
20
21
22 class DNSBruteforcer(BaseModule):
23 """Subdomain discovery via DNS brute force"""
24
25 MODULE_NAME = "dns_bruteforce"
26
27 def __init__(
28 self,
29 domain: str,
30 resolver: Optional['InternalDNSResolver'] = None,
31 wordlist_path: Optional[str] = None,
32 wordlist_size: str = 'medium',
33 threads: int = 30,
34 timeout: float = 2.0,
35 cancel_check: Optional[Callable[[], bool]] = None
36 ):
37 """
38 Initialize DNS brute forcer.
39
40 Args:
41 domain: Target domain
42 resolver: Custom DNS resolver (uses internal if None)
43 wordlist_path: Custom wordlist file path
44 wordlist_size: Size of bundled wordlist ('small', 'medium', 'large')
45 threads: Number of parallel threads
46 timeout: DNS query timeout
47 cancel_check: Optional function that returns True if cancelled
48 """
49 super().__init__(domain, resolver, timeout, cancel_check)
50
51 self.threads = threads
52 self.wordlist_path = wordlist_path
53 self.wordlist_size = wordlist_size
54 self.wordlist_manager = WordlistManager(wordlist_path)
55 self.wildcard_detector = WildcardDetector(domain, resolver)
56
57 def _resolve_subdomain(self, subdomain: str) -> tuple[str, Optional[str], Optional[Set[str]]]:
58 """
59 Resolve a single subdomain.
60
61 Args:
62 subdomain: Subdomain prefix to test
63
64 Returns:
65 Tuple of (full_subdomain, ip_address, all_ips)
66 """
67 fqdn = f"{subdomain}.{self.domain}"
68
69 ips = self.resolver.resolve_a_all(fqdn)
70 primary_ip = list(ips)[0] if ips else None
71
72 return fqdn, primary_ip, ips
73
74 def enumerate(
75 self,
76 max_words: Optional[int] = None,
77 progress_callback: Optional[Callable[[int, int, int], None]] = None
78 ) -> Dict[str, Optional[str]]:
79 """
80 Perform DNS brute force enumeration.
81
82 Args:
83 max_words: Maximum number of words to test
84 progress_callback: Callback for progress updates (current, total, found)
85
86 Returns:
87 Dict of {subdomain: ip_address}
88 """
89 if self.is_cancelled():
90 return self.discovered
91
92 logger.info(f"Starting DNS brute force for {self.domain}")
93
94 # Detect wildcards first
95 has_wildcard, wildcard_ips = self.wildcard_detector.detect()
96 if has_wildcard:
97 logger.warning(f"Wildcard DNS detected for {self.domain}: {wildcard_ips}")
98
99 if self.is_cancelled():
100 return self.discovered
101
102 # Load wordlist
103 wordlist = self.wordlist_manager.get_wordlist(
104 size=self.wordlist_size,
105 max_words=max_words,
106 include_numbers=True,
107 include_env=True
108 )
109
110 total = len(wordlist)
111 logger.info(f"Testing {total} subdomain candidates")
112
113 completed = 0
114 found = 0
115
116 with ThreadPoolExecutor(max_workers=self.threads) as executor:
117 futures = {
118 executor.submit(self._resolve_subdomain, word): word
119 for word in wordlist
120 }
121
122 for future in as_completed(futures):
123 # Check for cancellation
124 if self.is_cancelled():
125 # Cancel remaining futures and exit
126 for f in futures:
127 f.cancel()
128 executor.shutdown(wait=False, cancel_futures=True)
129 logger.info("DNS brute force cancelled")
130 return self.discovered
131
132 try:
133 fqdn, ip, all_ips = future.result(timeout=self.timeout + 1)
134 completed += 1
135
136 if ip:
137 # Check if it's a wildcard result
138 if has_wildcard and all_ips and self.wildcard_detector.is_wildcard_result(all_ips):
139 logger.debug(f"Filtered wildcard result: {fqdn}")
140 else:
141 self._add_discovered(fqdn, ip)
142 found += 1
143
144 # Progress update every 100 results
145 if progress_callback and completed % 100 == 0:
146 progress_callback(completed, total, found)
147
148 except Exception as e:
149 completed += 1
150 logger.debug(f"Error in brute force: {e}")
151
152 # Final progress update
153 if progress_callback:
154 progress_callback(completed, total, found)
155
156 logger.info(f"DNS brute force complete: {found} subdomains found")
157 return self.discovered
158
159 def test_specific(self, subdomains: Set[str]) -> Dict[str, Optional[str]]:
160 """
161 Test specific subdomain candidates.
162
163 Args:
164 subdomains: Set of subdomain prefixes to test
165
166 Returns:
167 Dict of {subdomain: ip_address}
168 """
169 results = {}
170
171 with ThreadPoolExecutor(max_workers=self.threads) as executor:
172 futures = {
173 executor.submit(self._resolve_subdomain, sub): sub
174 for sub in subdomains
175 }
176
177 for future in as_completed(futures):
178 try:
179 fqdn, ip, _ = future.result()
180 if ip:
181 results[fqdn] = ip
182 self._add_discovered(fqdn, ip)
183 except Exception as e:
184 logger.debug(f"Error testing subdomain: {e}")
185
186 return results