src/usfx/modules/dnssec_walker.py (5936B)
1 """
2 DNSSEC Zone Walking Module
3
4 Exploits DNSSEC NSEC/NSEC3 records to enumerate zone contents.
5 NSEC records form a chain linking all records in a zone.
6 """
7
8 import logging
9 from typing import Dict, Optional, Set, Tuple, TYPE_CHECKING
10
11 import dns.rdatatype
12
13 from .base import BaseModule
14
15 if TYPE_CHECKING:
16 from ..utils.dns_resolver import InternalDNSResolver
17
18 logger = logging.getLogger(__name__)
19
20
21 class DNSSECWalker(BaseModule):
22 """DNSSEC zone walking via NSEC/NSEC3 records"""
23
24 MODULE_NAME = "dnssec_walker"
25
26 def __init__(
27 self,
28 domain: str,
29 resolver: Optional['InternalDNSResolver'] = None,
30 timeout: float = 5.0,
31 max_iterations: int = 500
32 ):
33 """
34 Initialize DNSSEC walker.
35
36 Args:
37 domain: Target domain
38 resolver: Custom DNS resolver
39 timeout: DNS query timeout
40 max_iterations: Maximum NSEC chain iterations
41 """
42 super().__init__(domain, resolver, timeout)
43 self.max_iterations = max_iterations
44
45 def _check_dnssec(self) -> Tuple[bool, bool, dict]:
46 """
47 Check if domain uses DNSSEC and what type.
48
49 Returns:
50 Tuple of (has_dnssec, uses_nsec3, nsec3_params)
51 """
52 try:
53 # Try to get NSEC3PARAM first
54 records = self.resolver.resolve_any(self.domain, 'NSEC3PARAM')
55 if records:
56 # NSEC3 is in use
57 return True, True, {}
58
59 # Try to get DNSKEY to confirm DNSSEC
60 records = self.resolver.resolve_any(self.domain, 'DNSKEY')
61 if records:
62 return True, False, {}
63
64 return False, False, {}
65
66 except Exception as e:
67 logger.debug(f"Error checking DNSSEC: {e}")
68 return False, False, {}
69
70 def _walk_nsec(self) -> Set[str]:
71 """
72 Walk NSEC records to enumerate zone.
73
74 Returns:
75 Set of discovered subdomains
76 """
77 discovered = set()
78 current = self.domain
79 iterations = 0
80
81 logger.info(f"Starting NSEC zone walk for {self.domain}")
82
83 native_resolver = self.resolver.get_native_resolver()
84
85 while iterations < self.max_iterations:
86 iterations += 1
87
88 try:
89 # Query for NSEC record at current name
90 try:
91 answers = native_resolver.resolve(current, 'NSEC')
92 for rdata in answers:
93 next_name = str(rdata.next).rstrip('.')
94
95 if next_name.endswith(f".{self.domain}"):
96 discovered.add(next_name)
97 self._add_discovered(next_name)
98
99 # Check if we've looped back
100 if next_name == self.domain or next_name <= current:
101 logger.info(f"NSEC walk complete: {len(discovered)} found")
102 return discovered
103
104 current = next_name
105
106 except Exception:
107 # Try querying a fake subdomain to get NSEC in authority
108 fake = f"aaa{iterations}.{current if current != self.domain else self.domain}"
109 try:
110 native_resolver.resolve(fake, 'A')
111 except Exception:
112 break
113
114 except Exception as e:
115 logger.debug(f"NSEC walk error at {current}: {e}")
116 break
117
118 logger.info(f"NSEC walk finished with {len(discovered)} subdomains")
119 return discovered
120
121 def _walk_nsec3(self, params: dict) -> Set[str]:
122 """
123 Attempt to crack NSEC3 hashes (limited effectiveness).
124
125 For NSEC3, we can only try to match known names against hashes,
126 so this is less effective than NSEC walking.
127
128 Args:
129 params: NSEC3 parameters
130
131 Returns:
132 Set of discovered subdomains
133 """
134 discovered = set()
135
136 # NSEC3 uses hashed names, making direct walking impossible
137 # We can only verify if specific names exist by computing their hashes
138
139 # Common subdomain prefixes to try
140 common_prefixes = [
141 'www', 'mail', 'ftp', 'smtp', 'pop', 'imap', 'webmail',
142 'api', 'dev', 'test', 'staging', 'admin', 'blog', 'shop',
143 'cdn', 'static', 'ns1', 'ns2', 'dns', 'vpn', 'remote'
144 ]
145
146 logger.info(f"Attempting NSEC3 hash matching for {self.domain}")
147
148 for prefix in common_prefixes:
149 subdomain = f"{prefix}.{self.domain}"
150 ip = self.resolver.resolve_a(subdomain)
151 if ip:
152 discovered.add(subdomain)
153 self._add_discovered(subdomain, ip)
154
155 logger.info(f"NSEC3 matching found {len(discovered)} subdomains")
156 return discovered
157
158 def enumerate(self) -> Dict[str, Optional[str]]:
159 """
160 Enumerate subdomains using DNSSEC zone walking.
161
162 Returns:
163 Dict of {subdomain: ip_address}
164 """
165 logger.info(f"Checking DNSSEC for {self.domain}")
166
167 has_dnssec, uses_nsec3, nsec3_params = self._check_dnssec()
168
169 if not has_dnssec:
170 logger.info(f"Domain {self.domain} does not use DNSSEC")
171 return {}
172
173 if uses_nsec3:
174 logger.info(f"Domain uses NSEC3 (hashed) - limited enumeration possible")
175 self._walk_nsec3(nsec3_params)
176 else:
177 logger.info(f"Domain uses NSEC - zone walking possible")
178 self._walk_nsec()
179
180 # Resolve discovered subdomains to get IPs
181 for subdomain in list(self.discovered.keys()):
182 if self.discovered[subdomain] is None:
183 ip = self.resolver.resolve_a(subdomain)
184 if ip:
185 self.discovered[subdomain] = ip
186
187 return self.discovered