315 lines
13 KiB
Python
315 lines
13 KiB
Python
"""DNS leak detection and cache flushing.
|
|
|
|
Two levels of testing:
|
|
1. check_dns_leak_hint() — fast, no network: inspect configured DNS resolvers
|
|
2. run_dns_leak_test() — real network test: compare DNS answers seen through
|
|
the proxy vs direct, and probe resolver identity via
|
|
a dedicated leak-test API.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import concurrent.futures
|
|
import ipaddress
|
|
import logging
|
|
import socket
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
|
|
import httpx
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ─── DoH endpoint used to ask "what is my apparent source IP?" ──────────────
|
|
_DOH_IP_CHECK = "https://api.ipify.org?format=json"
|
|
_DNS_LEAK_API = "https://dnsleaktest.com/api/v1/start"
|
|
_DNS_LEAK_RESULT = "https://dnsleaktest.com/api/v1/results"
|
|
|
|
# Well-known public resolver IP ranges (prefix → label)
|
|
_PUBLIC_DNS_LABELS: dict[str, str] = {
|
|
"8.8.8.8": "Google", "8.8.4.4": "Google",
|
|
"1.1.1.1": "Cloudflare", "1.0.0.1": "Cloudflare",
|
|
"9.9.9.9": "Quad9", "149.112.112.112": "Quad9",
|
|
"208.67.222.222": "OpenDNS", "208.67.220.220": "OpenDNS",
|
|
"76.76.2.0": "Alternate DNS",
|
|
"94.140.14.14": "AdGuard",
|
|
}
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
@dataclass
|
|
class DnsLeakResult:
|
|
"""Legacy single-check result kept for API compatibility."""
|
|
ok: bool
|
|
system_resolvers: list[str]
|
|
message: str
|
|
doh_ip: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class DnsResolverInfo:
|
|
ip: str
|
|
hostname: str = ""
|
|
country: str = ""
|
|
isp: str = ""
|
|
label: str = "" # "Google", "Cloudflare", or ""
|
|
is_public_known: bool = False
|
|
|
|
|
|
@dataclass
|
|
class FullDnsLeakReport:
|
|
"""Comprehensive DNS leak report from run_dns_leak_test()."""
|
|
# Resolvers that answered DNS queries in this session
|
|
resolvers_seen: list[DnsResolverInfo] = field(default_factory=list)
|
|
# Resolvers configured by the OS
|
|
system_resolvers: list[str] = field(default_factory=list)
|
|
# IPs resolved for test hostnames — via proxy vs direct
|
|
proxy_resolution: dict[str, list[str]] = field(default_factory=dict)
|
|
direct_resolution: dict[str, list[str]] = field(default_factory=dict)
|
|
# True when proxy and direct give the same answers (possible leak)
|
|
resolution_matches_direct: bool = False
|
|
# Whether any configured resolver is a known public server
|
|
has_public_resolver: bool = False
|
|
# Whether any configured resolver is outside local network
|
|
has_external_resolver: bool = False
|
|
# Overall verdict
|
|
leaked: bool = False
|
|
summary: str = ""
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# System resolver detection
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def get_system_dns_servers() -> list[str]:
|
|
"""Return configured IPv4 DNS resolvers via PowerShell (falls back to ipconfig)."""
|
|
try:
|
|
r = subprocess.run(
|
|
["powershell", "-NoProfile", "-Command",
|
|
"(Get-DnsClientServerAddress -AddressFamily IPv4 | "
|
|
"Where-Object { $_.ServerAddresses } | "
|
|
"Select-Object -ExpandProperty ServerAddresses) -join ','"],
|
|
capture_output=True, text=True, timeout=12,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
raw = (r.stdout or "").strip()
|
|
if raw:
|
|
return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()]
|
|
except Exception as exc:
|
|
log.debug("PowerShell DNS server query failed (will try ipconfig): %s", exc)
|
|
|
|
# ipconfig fallback
|
|
try:
|
|
r = subprocess.run(
|
|
["ipconfig", "/all"], capture_output=True, text=True, timeout=10,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
servers: list[str] = []
|
|
for line in (r.stdout or "").splitlines():
|
|
if "DNS Servers" in line or "DNS Server" in line:
|
|
parts = line.split(":", 1)
|
|
if len(parts) == 2:
|
|
ip = parts[1].strip()
|
|
if ip:
|
|
servers.append(ip)
|
|
return servers
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def flush_dns_cache() -> tuple[bool, str]:
|
|
try:
|
|
r = subprocess.run(
|
|
["ipconfig", "/flushdns"], capture_output=True, text=True, timeout=15,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
lines = (r.stdout or r.stderr or "").strip().splitlines()
|
|
msg = lines[-1] if lines else "flushed"
|
|
return r.returncode == 0, msg
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Hostname resolution helpers
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _resolve_direct(hostname: str, timeout: float = 4.0) -> list[str]:
|
|
"""Resolve hostname using system DNS (direct, not through proxy)."""
|
|
_prev = socket.getdefaulttimeout()
|
|
try:
|
|
socket.setdefaulttimeout(timeout)
|
|
infos = socket.getaddrinfo(hostname, None)
|
|
seen: list[str] = []
|
|
for info in infos:
|
|
addr = info[4][0]
|
|
if addr and addr not in seen:
|
|
seen.append(addr)
|
|
return seen
|
|
except Exception:
|
|
return []
|
|
finally:
|
|
socket.setdefaulttimeout(_prev)
|
|
|
|
|
|
def _resolve_via_proxy(hostname: str, proxy_url: str, timeout: float = 8.0) -> list[str]:
|
|
"""Resolve hostname by fetching a DNS-over-HTTPS endpoint through the proxy."""
|
|
# Use Google DoH JSON API through the proxy so DNS is resolved on the exit node
|
|
doh = f"https://dns.google/resolve?name={hostname}&type=A"
|
|
try:
|
|
with httpx.Client(proxy=proxy_url, timeout=timeout, verify=False, follow_redirects=True) as c:
|
|
r = c.get(doh)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
return [ans["data"] for ans in (data.get("Answer") or [])
|
|
if ans.get("type") == 1]
|
|
except Exception as exc:
|
|
log.debug("DoH DNS leak query failed: %s", exc)
|
|
return []
|
|
|
|
|
|
def _is_private(ip: str) -> bool:
|
|
try:
|
|
return ipaddress.ip_address(ip).is_private
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Main leak test
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
_LEAK_TEST_HOSTS = [
|
|
"google.com",
|
|
"cloudflare.com",
|
|
"github.com",
|
|
"amazon.com",
|
|
]
|
|
|
|
|
|
def run_dns_leak_test(
|
|
proxy_url: str | None = None,
|
|
timeout: float = 10.0,
|
|
) -> FullDnsLeakReport:
|
|
"""
|
|
Comprehensive DNS leak test:
|
|
- Collect system DNS resolver configuration
|
|
- Resolve test hostnames both directly (system DNS) and via proxy (DoH through chain)
|
|
- Compare answers: matching answers suggest DNS is NOT going through the proxy
|
|
- Flag known public resolvers that would bypass the chain
|
|
"""
|
|
rep = FullDnsLeakReport()
|
|
rep.system_resolvers = get_system_dns_servers()
|
|
|
|
_PRIVATE_PREFIXES = (
|
|
"127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.",
|
|
"172.19.", "172.2", "172.30.", "172.31.", "169.254.",
|
|
)
|
|
|
|
for r_ip in rep.system_resolvers:
|
|
is_public_known = r_ip in _PUBLIC_DNS_LABELS
|
|
is_external = not any(r_ip.startswith(p) for p in _PRIVATE_PREFIXES)
|
|
label = _PUBLIC_DNS_LABELS.get(r_ip, "")
|
|
rep.has_public_resolver = rep.has_public_resolver or is_public_known
|
|
rep.has_external_resolver = rep.has_external_resolver or is_external
|
|
rep.resolvers_seen.append(DnsResolverInfo(
|
|
ip=r_ip,
|
|
label=label,
|
|
is_public_known=is_public_known,
|
|
))
|
|
|
|
# Resolve test hosts in parallel
|
|
all_match = True
|
|
any_resolved_via_proxy = False
|
|
|
|
def _test_host(host: str) -> tuple[str, list[str], list[str]]:
|
|
direct = _resolve_direct(host, timeout=min(4.0, timeout))
|
|
proxy = _resolve_via_proxy(host, proxy_url, timeout=timeout) if proxy_url else []
|
|
return host, direct, proxy
|
|
|
|
try:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
|
|
futs = {pool.submit(_test_host, h): h for h in _LEAK_TEST_HOSTS}
|
|
for fut in concurrent.futures.as_completed(futs, timeout=timeout + 2):
|
|
try:
|
|
host, direct, proxy = fut.result()
|
|
rep.direct_resolution[host] = direct
|
|
rep.proxy_resolution[host] = proxy
|
|
if proxy:
|
|
any_resolved_via_proxy = True
|
|
# Overlap: if proxy and direct gave the same IPs, DNS
|
|
# might not be going through the proxy (same resolver path)
|
|
direct_set = set(direct)
|
|
proxy_set = set(proxy)
|
|
if direct_set and proxy_set and direct_set == proxy_set:
|
|
pass # this host matches
|
|
else:
|
|
all_match = False
|
|
except Exception as e:
|
|
rep.errors.append(str(e))
|
|
except Exception as e:
|
|
rep.errors.append(f"Thread pool error: {e}")
|
|
|
|
rep.resolution_matches_direct = all_match and any_resolved_via_proxy
|
|
|
|
# ── Build verdict ──────────────────────────────────────────────────────
|
|
problems: list[str] = []
|
|
|
|
if rep.has_external_resolver:
|
|
external = [r for r in rep.system_resolvers
|
|
if not any(r.startswith(p) for p in _PRIVATE_PREFIXES)]
|
|
names = [f"{ip} ({_PUBLIC_DNS_LABELS[ip]})" if ip in _PUBLIC_DNS_LABELS else ip
|
|
for ip in external[:4]]
|
|
problems.append(f"OS DNS: {', '.join(names)} — DNS may bypass proxy chain")
|
|
|
|
if rep.resolution_matches_direct and proxy_url:
|
|
problems.append("Proxy DNS resolution matches direct — possible DNS leak (same upstream resolver)")
|
|
|
|
if problems:
|
|
rep.leaked = True
|
|
rep.summary = " · ".join(problems)
|
|
else:
|
|
rep.leaked = False
|
|
if proxy_url:
|
|
rep.summary = "DNS routing looks clean — proxy resolves differently from direct."
|
|
else:
|
|
resolver_str = ", ".join(rep.system_resolvers[:3]) if rep.system_resolvers else "none detected"
|
|
rep.summary = f"System resolvers: {resolver_str}"
|
|
|
|
return rep
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Legacy quick hint (used by Privacy tab DNS section)
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def check_dns_leak_hint(local_proxy: str | None = None) -> DnsLeakResult:
|
|
"""Fast heuristic: inspect configured DNS resolvers, flag public ones."""
|
|
resolvers = get_system_dns_servers()
|
|
private_prefixes = (
|
|
"127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.",
|
|
"172.19.", "172.2", "172.30.", "172.31.",
|
|
)
|
|
public = [r for r in resolvers if not any(r.startswith(p) for p in private_prefixes)]
|
|
|
|
if not resolvers:
|
|
return DnsLeakResult(ok=True, system_resolvers=[], message="No IPv4 DNS servers reported (DHCP).")
|
|
|
|
if public and local_proxy:
|
|
return DnsLeakResult(
|
|
ok=False, system_resolvers=resolvers,
|
|
message=(
|
|
f"DNS may bypass proxy chain: public resolvers {', '.join(public)}. "
|
|
"Use kill-switch + VPN, or set DNS to localhost when hardened."
|
|
),
|
|
)
|
|
if public:
|
|
return DnsLeakResult(
|
|
ok=False, system_resolvers=resolvers,
|
|
message=f"Public DNS resolvers active: {', '.join(public)}",
|
|
)
|
|
return DnsLeakResult(
|
|
ok=True, system_resolvers=resolvers,
|
|
message=f"DNS servers: {', '.join(resolvers)}",
|
|
)
|