Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 15,037 Bytes
d543fc1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 | """
dns_security_scanner.py — DNS security analysis using DNSPython.
Performs comprehensive DNS security checks including zone transfer, DNSSEC, and record analysis.
"""
import dns.resolver, dns.query, dns.message, dns.rdatatype
from scanners.base_scanner import BaseScanner
class DNSSecurityScanner(BaseScanner):
SCANNER_NAME = "DNS Security Analyzer"
def run(self):
self.log("INFO", f"[DNS Security] Starting comprehensive DNS analysis for {self.domain}...")
# Perform various DNS security checks
self._check_zone_transfer()
self._check_dnssec()
self._check_mx_records()
self._check_spf()
self._check_dmarc()
self._check_wildcard_dns()
self._check_dns_cache_poisoning()
self._check_caa_records()
self.log("SUCCESS", "[DNS Security] DNS analysis complete.")
return self.vulns
def _check_zone_transfer(self):
"""Check for DNS zone transfer vulnerability."""
self.log("INFO", "[DNS Security] Checking for zone transfer vulnerability...")
try:
# Get authoritative nameservers
ns_answers = dns.resolver.resolve(self.domain, 'NS')
nameservers = [rdata.target.to_text().rstrip('.') for rdata in ns_answers]
for ns in nameservers[:2]: # Check first 2 nameservers only (faster)
try:
# Attempt AXFR (zone transfer) — short timeout for speed
zone_query = dns.query.xfr(ns, self.domain, timeout=5)
zone_response = next(zone_query)
if zone_response:
self.log("CRITICAL", f"[DNS Security] ZONE TRANSFER VULNERABLE on {ns}!")
self.add_vuln(
title="DNS Zone Transfer Vulnerability",
severity="Critical", category="DNS Security", cvss_score=9.8,
description=f"DNS zone transfer (AXFR) is allowed from {ns}. This exposes all DNS records including internal hostnames, mail servers, and other sensitive infrastructure details.",
remediation="Disable zone transfers to unauthorized servers:\n BIND: allow-transfer { trusted_ips; };\n Ensure only authorized secondary nameservers can perform AXFR."
)
return # Found vulnerability, no need to check others
except dns.exception.DNSException:
# Expected if zone transfer is disabled
pass
except OSError as e:
# Network-level errors (timeout, connection refused) are expected
err_msg = str(e).strip()
if err_msg:
self.log("DEBUG", f"[DNS Security] Zone transfer network error on {ns}: {err_msg}")
except Exception as e:
err_msg = str(e).strip()
if err_msg:
self.log("DEBUG", f"[DNS Security] Nameserver query error on {ns}: {err_msg}")
self.log("SUCCESS", "[DNS Security] Zone transfer properly secured.")
except dns.resolver.NoNameservers:
self.log("WARNING", "[DNS Security] Could not retrieve nameservers.")
except Exception as e:
self.log("WARNING", f"[DNS Security] Zone transfer check failed: {e}")
def _check_dnssec(self):
"""Check for DNSSEC implementation."""
self.log("INFO", "[DNS Security] Checking DNSSEC implementation...")
try:
# Check for DNSKEY records
dnskey_answers = dns.resolver.resolve(self.domain, 'DNSKEY')
if dnskey_answers:
self.log("SUCCESS", "[DNS Security] DNSSEC is implemented.")
return
except dns.resolver.NoAnswer:
pass
except Exception as e:
self.log("ERROR", f"[DNS Security] DNSKEY query error: {e}")
self.log("WARNING", "[DNS Security] DNSSEC not implemented.")
self.add_vuln(
title="DNSSEC Not Implemented",
severity="Medium", category="DNS Security", cvss_score=5.3,
description=f"The domain {self.domain} does not have DNSSEC implemented. This makes it vulnerable to DNS cache poisoning and spoofing attacks.",
remediation="Implement DNSSEC:\n 1. Generate DNSSEC keys\n 2. Sign your zone files\n 3. Upload DS records to your registrar\n 4. Verify with: dig +dnssec {self.domain}"
)
def _check_mx_records(self):
"""Check MX records for security issues."""
self.log("INFO", "[DNS Security] Analyzing MX records...")
try:
mx_answers = dns.resolver.resolve(self.domain, 'MX')
mx_servers = [rdata.exchange.to_text().rstrip('.') for rdata in mx_answers]
if mx_servers:
self.log("SUCCESS", f"[DNS Security] MX Records: {', '.join(mx_servers[:3])}")
# Check for common mail server security issues
for mx in mx_servers:
# Check if MX points to IP directly (bad practice)
if mx.replace('.', '').isdigit():
self.log("WARNING", f"[DNS Security] MX record points to IP: {mx}")
self.add_vuln(
title="MX Record Points to IP Address",
severity="Low", category="DNS Security", cvss_score=3.1,
description=f"MX record for {self.domain} points to IP address {mx} instead of a hostname. This is not a best practice.",
remediation="Update MX records to point to hostnames instead of IP addresses."
)
else:
self.log("INFO", "[DNS Security] No MX records found.")
except dns.resolver.NoAnswer:
self.log("INFO", "[DNS Security] No MX records found.")
except Exception as e:
self.log("WARNING", f"[DNS Security] MX check failed: {e}")
def _check_spf(self):
"""Check for SPF record."""
self.log("INFO", "[DNS Security] Checking SPF record...")
try:
txt_answers = dns.resolver.resolve(self.domain, 'TXT')
spf_found = False
for rdata in txt_answers:
txt = rdata.to_text()
if 'v=spf1' in txt:
spf_found = True
self.log("SUCCESS", f"[DNS Security] SPF Record: {txt[:100]}")
# Check for ~all vs -all
if '~all' in txt:
self.log("WARNING", "[DNS Security] SPF uses ~all (soft fail)")
self.add_vuln(
title="SPF Uses Soft Fail (~all)",
severity="Low", category="DNS Security", cvss_score=3.1,
description=f"SPF record for {self.domain} uses ~all (soft fail) instead of -all (hard fail). This may allow some spoofed emails to pass.",
remediation="Consider changing ~all to -all in your SPF record for stricter email validation."
)
break
if not spf_found:
self.log("WARNING", "[DNS Security] No SPF record found.")
self.add_vuln(
title="SPF Record Missing",
severity="Medium", category="DNS Security", cvss_score=5.3,
description=f"No SPF (Sender Policy Framework) record found for {self.domain}. This allows email spoofing and increases spam risk.",
remediation="Add an SPF record to your DNS:\n Example: v=spf1 ip4:192.0.2.0/24 -all\n Use SPF record generator tools for proper configuration."
)
except dns.resolver.NoAnswer:
self.log("WARNING", "[DNS Security] No TXT records found (SPF missing).")
self.add_vuln(
title="SPF Record Missing",
severity="Medium", category="DNS Security", cvss_score=5.3,
description=f"No SPF record found for {self.domain}.",
remediation="Add an SPF record to your DNS configuration."
)
except Exception as e:
self.log("WARNING", f"[DNS Security] SPF check failed: {e}")
def _check_dmarc(self):
"""Check for DMARC record."""
self.log("INFO", "[DNS Security] Checking DMARC record...")
dmarc_domain = f"_dmarc.{self.domain}"
try:
txt_answers = dns.resolver.resolve(dmarc_domain, 'TXT')
dmarc_found = False
for rdata in txt_answers:
txt = rdata.to_text()
if 'v=DMARC1' in txt:
dmarc_found = True
self.log("SUCCESS", f"[DNS Security] DMARC Record: {txt[:100]}")
# Check for p=none (monitoring only)
if 'p=none' in txt:
self.log("WARNING", "[DNS Security] DMARC policy is p=none (monitoring only)")
self.add_vuln(
title="DMARC Policy Set to None",
severity="Low", category="DNS Security", cvss_score=3.1,
description=f"DMARC record for {self.domain} has policy p=none, which means no action is taken on failed SPF/DKIM. This is only suitable for monitoring.",
remediation="Update DMARC policy to p=quarantine or p=reject for better email security:\n v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com"
)
break
if not dmarc_found:
self.log("WARNING", "[DNS Security] No DMARC record found.")
self.add_vuln(
title="DMARC Record Missing",
severity="Medium", category="DNS Security", cvss_score=5.3,
description=f"No DMARC (Domain-based Message Authentication, Reporting & Conformance) record found for {self.domain}. DMARC helps prevent email spoofing.",
remediation="Add a DMARC record to your DNS:\n Example: v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com\n Start with p=none for monitoring, then move to p=quarantine or p=reject."
)
except dns.resolver.NoAnswer:
self.log("WARNING", "[DNS Security] No DMARC record found.")
self.add_vuln(
title="DMARC Record Missing",
severity="Medium", category="DNS Security", cvss_score=5.3,
description=f"No DMARC record found for {self.domain}.",
remediation="Add a DMARC record to your DNS configuration."
)
except dns.resolver.NXDOMAIN:
self.log("WARNING", "[DNS Security] DMARC subdomain does not exist.")
except Exception as e:
self.log("WARNING", f"[DNS Security] DMARC check failed: {e}")
def _check_wildcard_dns(self):
"""Check for wildcard DNS records."""
self.log("INFO", "[DNS Security] Checking for wildcard DNS...")
random_subdomain = f"random-nonexistent-{hash(self.domain) % 10000}.{self.domain}"
try:
dns.resolver.resolve(random_subdomain, 'A')
self.log("WARNING", "[DNS Security] Wildcard DNS record detected.")
self.add_vuln(
title="Wildcard DNS Record Detected",
severity="Low", category="DNS Security", cvss_score=3.1,
description=f"Wildcard DNS record detected for {self.domain}. This can lead to subdomain takeover vulnerabilities if not properly managed.",
remediation="Review wildcard DNS usage. Ensure all subdomains are properly claimed and monitored. Consider removing wildcard if not necessary."
)
except dns.resolver.NXDOMAIN:
self.log("SUCCESS", "[DNS Security] No wildcard DNS detected.")
except Exception as e:
self.log("WARNING", f"[DNS Security] Wildcard check failed: {e}")
def _check_dns_cache_poisoning(self):
"""Check for DNS cache poisoning vulnerabilities."""
self.log("INFO", "[DNS Security] Checking DNS cache poisoning risks...")
try:
# Check for random source ports (good practice)
# This is a heuristic check - we can't directly test this without more complex probing
self.log("INFO", "[DNS Security] Ensure DNS resolver uses random source ports and transaction IDs.")
# Check for DNSSEC (already done in _check_dnssec)
# This is informational
self.log("INFO", "[DNS Security] DNS cache poisoning protection relies on DNSSEC implementation.")
except Exception as e:
self.log("WARNING", f"[DNS Security] Cache poisoning check failed: {e}")
def _check_caa_records(self):
"""Check for CAA (Certification Authority Authorization) records."""
self.log("INFO", "[DNS Security] Checking CAA records...")
try:
caa_answers = dns.resolver.resolve(self.domain, 'CAA')
caa_records = []
for rdata in caa_answers:
caa_records.append(f"{rdata.flags} {rdata.tag} {rdata.value}")
if caa_records:
self.log("SUCCESS", f"[DNS Security] CAA Records: {', '.join(caa_records)}")
else:
self.log("INFO", "[DNS Security] No CAA records found (informational).")
self.add_vuln(
title="CAA Record Not Configured",
severity="Low", category="DNS Security", cvss_score=2.1,
description=f"No CAA (Certification Authority Authorization) record found for {self.domain}. CAA records specify which CAs are allowed to issue certificates for the domain.",
remediation="Consider adding CAA records to restrict which certificate authorities can issue certificates:\n Example: issue ca.example.com; issuewild ca.example.com"
)
except dns.resolver.NoAnswer:
self.log("INFO", "[DNS Security] No CAA records found.")
except Exception as e:
self.log("WARNING", f"[DNS Security] CAA check failed: {e}")
|