340 lines
12 KiB
Python
340 lines
12 KiB
Python
"""Consolidated live leak audit.
|
|
|
|
Used by the Privacy tab "Run audit" button. Probes every surface that can
|
|
deanonymize the host even when the chain is healthy:
|
|
|
|
• IP — exit IP through chain vs direct IP (subnet leak)
|
|
• DNS — system resolvers, whether they're public
|
|
• IPv6 binding — adapters with IPv6 active
|
|
• WPAD/PAC — leftover AutoConfigURL or AutoDetect
|
|
• Policy locks — Group Policy proxy keys that beat our settings
|
|
• PerUser flag — Windows Server ProxySettingsPerUser
|
|
• LAN broadcast — LLMNR / NetBIOS / mDNS status
|
|
• VPN — adapter presence
|
|
• WebRTC — Chrome/Edge policy presence
|
|
• System proxy — what registry says we're set to
|
|
• TLS exit — quick categorisation hint (datacenter / residential)
|
|
|
|
Each check has its own short timeout so a slow probe never blocks the rest.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import subprocess
|
|
import sys
|
|
import winreg
|
|
from dataclasses import dataclass, field
|
|
|
|
from .dns_leak import get_system_dns_servers
|
|
from .firewall import is_admin
|
|
from .privacy_lan import lan_status
|
|
from .sysproxy import detect_policy_overrides, is_system_proxy_set
|
|
from .validator import check_chain_exit_ip, get_direct_ip
|
|
from .vpn_detect import detect_vpn
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class AuditFinding:
|
|
name: str
|
|
ok: bool
|
|
value: str
|
|
note: str = ""
|
|
|
|
|
|
@dataclass
|
|
class AuditReport:
|
|
findings: list[AuditFinding] = field(default_factory=list)
|
|
overall_ok: bool = True
|
|
|
|
def add(self, name: str, ok: bool, value: str, note: str = "") -> None:
|
|
self.findings.append(AuditFinding(name, ok, value, note))
|
|
if not ok:
|
|
self.overall_ok = False
|
|
|
|
|
|
def _check_ipv6_active() -> tuple[bool, str]:
|
|
"""True/'string' if any 'Up' adapter has IPv6 binding enabled.
|
|
|
|
PowerShell 3.0+ path; falls back to ipconfig parsing (locale-tolerant
|
|
enough — looks for hex colons).
|
|
"""
|
|
if sys.platform == "darwin":
|
|
try:
|
|
r = subprocess.run(["ifconfig"], capture_output=True, text=True, timeout=8)
|
|
active: list[str] = []
|
|
current = ""
|
|
for line in (r.stdout or "").splitlines():
|
|
if line and not line.startswith("\t") and ":" in line:
|
|
current = line.split(":", 1)[0]
|
|
elif "\tinet6 " in line and current and not line.strip().startswith("inet6 ::1"):
|
|
active.append(current)
|
|
uniq = sorted(set(active))
|
|
if uniq:
|
|
return True, ", ".join(uniq[:3]) + (f" +{len(uniq)-3}" if len(uniq) > 3 else "")
|
|
return False, "no non-loopback IPv6 addresses"
|
|
except Exception as e:
|
|
return False, f"unknown ({e})"
|
|
try:
|
|
r = subprocess.run(
|
|
["powershell", "-NoProfile", "-NonInteractive", "-Command",
|
|
"Get-NetAdapterBinding -ComponentID ms_tcpip6 | "
|
|
"Where-Object { $_.Enabled } | "
|
|
"Select-Object -ExpandProperty Name"],
|
|
capture_output=True, text=True, timeout=10,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
names = [n.strip() for n in (r.stdout or "").splitlines() if n.strip()]
|
|
if names:
|
|
return True, ", ".join(names[:3]) + (f" +{len(names)-3}" if len(names) > 3 else "")
|
|
# Empty stdout could mean PS3 cmdlet missing — try ipconfig fallback.
|
|
if r.returncode != 0:
|
|
raise RuntimeError("PS3 net cmdlets unavailable")
|
|
return False, "no adapters bound to IPv6"
|
|
except Exception:
|
|
try:
|
|
r = subprocess.run(
|
|
["ipconfig"], capture_output=True, text=True, timeout=8,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
has_v6 = any(
|
|
"IPv6" in ln and ":" in ln.split(":", 1)[1]
|
|
for ln in (r.stdout or "").splitlines()
|
|
)
|
|
return has_v6, ("IPv6 address present" if has_v6 else "no IPv6 address")
|
|
except Exception as e:
|
|
return False, f"unknown ({e})"
|
|
|
|
|
|
def _check_wpad() -> tuple[bool, str]:
|
|
if sys.platform != "win32":
|
|
return True, "not applicable on macOS"
|
|
path = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
|
|
autoconf = ""
|
|
autodet = 0
|
|
try:
|
|
with winreg.OpenKey(
|
|
winreg.HKEY_CURRENT_USER, path, 0, winreg.KEY_QUERY_VALUE
|
|
) as key:
|
|
try:
|
|
autoconf = str(winreg.QueryValueEx(key, "AutoConfigURL")[0] or "")
|
|
except OSError:
|
|
pass
|
|
try:
|
|
autodet = int(winreg.QueryValueEx(key, "AutoDetect")[0])
|
|
except OSError:
|
|
autodet = 0
|
|
except OSError:
|
|
pass
|
|
leaking = bool(autoconf) or autodet == 1
|
|
if leaking:
|
|
bits = []
|
|
if autoconf:
|
|
bits.append(f"AutoConfigURL={autoconf}")
|
|
if autodet:
|
|
bits.append(f"AutoDetect={autodet}")
|
|
return False, ", ".join(bits)
|
|
return True, "no WPAD/PAC override"
|
|
|
|
|
|
def _check_per_user_flag() -> tuple[bool, str]:
|
|
if sys.platform != "win32":
|
|
return True, "not applicable on macOS"
|
|
try:
|
|
with winreg.OpenKey(
|
|
winreg.HKEY_LOCAL_MACHINE,
|
|
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Internet Settings",
|
|
0, winreg.KEY_QUERY_VALUE,
|
|
) as key:
|
|
v = int(winreg.QueryValueEx(key, "ProxySettingsPerUser")[0])
|
|
if v == 0:
|
|
return False, "ProxySettingsPerUser=0 (HKCU IGNORED)"
|
|
return True, "ProxySettingsPerUser=1"
|
|
except OSError:
|
|
return True, "unset (default → HKCU honored)"
|
|
|
|
|
|
def _check_webrtc_policy() -> tuple[bool, str]:
|
|
"""OK means the policy IS set (browsers won't leak non-proxied UDP)."""
|
|
if sys.platform != "win32":
|
|
return True, "managed by hardened Firefox profile settings"
|
|
paths = (
|
|
r"SOFTWARE\Policies\Google\Chrome",
|
|
r"SOFTWARE\Policies\Microsoft\Edge",
|
|
)
|
|
_DWORD_DISABLE = 3 # disable_non_proxied_udp
|
|
_STR_KEY = "WebRtcIPHandling"
|
|
_STR_DISABLE = "disable_non_proxied_udp"
|
|
found = []
|
|
for p in paths:
|
|
try:
|
|
with winreg.OpenKey(
|
|
winreg.HKEY_LOCAL_MACHINE, p, 0, winreg.KEY_QUERY_VALUE
|
|
) as key:
|
|
ok = False
|
|
try:
|
|
v = int(winreg.QueryValueEx(key, "DefaultWebRtcIpHandlingPolicy")[0])
|
|
if v == _DWORD_DISABLE:
|
|
ok = True
|
|
except OSError:
|
|
pass
|
|
try:
|
|
v_str = str(winreg.QueryValueEx(key, _STR_KEY)[0]).strip().lower()
|
|
if v_str == _STR_DISABLE:
|
|
ok = True
|
|
except OSError:
|
|
pass
|
|
if ok:
|
|
found.append(p.split("\\")[-1])
|
|
except OSError:
|
|
continue
|
|
if found:
|
|
return True, "policy set: " + ", ".join(found)
|
|
return False, "no WebRTC policy — Chrome/Edge may leak local IPs"
|
|
|
|
|
|
def _categorize_exit(ip: str | None) -> str:
|
|
"""Cheap categorization hint based on common RIR / ASN heuristics.
|
|
|
|
No external lookup — just shape-based hints. Real classification can be
|
|
added later via offline GeoIP DBs.
|
|
"""
|
|
if not ip:
|
|
return "unknown"
|
|
try:
|
|
a = int(ip.split(".")[0])
|
|
except Exception:
|
|
return "non-IPv4"
|
|
if a in (10,) or ip.startswith(("192.168.", "172.16.", "172.17.", "172.18.",
|
|
"172.19.", "172.2", "172.30.", "172.31.")):
|
|
return "PRIVATE — chain broken"
|
|
if a == 127:
|
|
return "LOOPBACK — chain broken"
|
|
return "public"
|
|
|
|
|
|
async def run_audit(
|
|
listen_proxy: str,
|
|
ip_check_url: str,
|
|
timeout_seconds: float = 10.0,
|
|
) -> AuditReport:
|
|
"""Run every check in parallel where safe; return structured report."""
|
|
rep = AuditReport()
|
|
chain_running = is_system_proxy_set()
|
|
|
|
# Network probes in parallel
|
|
direct_task = asyncio.create_task(get_direct_ip(ip_check_url, timeout_seconds))
|
|
chain_task = (
|
|
asyncio.create_task(
|
|
check_chain_exit_ip(listen_proxy, ip_check_url, timeout_seconds, chain_hops=1)
|
|
)
|
|
if chain_running else None
|
|
)
|
|
|
|
direct_ip = await direct_task
|
|
exit_ip = await chain_task if chain_task else None
|
|
|
|
if direct_ip:
|
|
rep.add("Direct IP (no chain)", True, direct_ip, "")
|
|
else:
|
|
rep.add("Direct IP (no chain)", False, "unreachable", "no internet?")
|
|
|
|
if chain_running:
|
|
if exit_ip:
|
|
cat = _categorize_exit(exit_ip)
|
|
same = bool(direct_ip and exit_ip == direct_ip)
|
|
rep.add(
|
|
"Chain exit IP",
|
|
not same and cat == "public",
|
|
f"{exit_ip} ({cat})",
|
|
"matches direct IP — chain not forwarding" if same else "",
|
|
)
|
|
else:
|
|
rep.add("Chain exit IP", False, "unreachable through chain", "")
|
|
else:
|
|
rep.add("Chain exit IP", True, "chain not running", "skipped")
|
|
|
|
# DNS resolvers
|
|
dns_servers = 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_dns = [d for d in dns_servers if not d.startswith(private_prefixes)]
|
|
if not dns_servers:
|
|
rep.add("DNS resolvers", True, "none reported (DHCP)", "")
|
|
elif public_dns:
|
|
rep.add(
|
|
"DNS resolvers",
|
|
False,
|
|
", ".join(dns_servers),
|
|
f"{len(public_dns)} public — DNS may bypass chain",
|
|
)
|
|
else:
|
|
rep.add("DNS resolvers", True, ", ".join(dns_servers), "all private/local")
|
|
|
|
# IPv6 binding
|
|
v6_on, v6_msg = _check_ipv6_active()
|
|
rep.add("IPv6 binding", not v6_on, v6_msg, "IPv6 active = potential leak past v4 proxies" if v6_on else "")
|
|
|
|
# WPAD
|
|
wpad_ok, wpad_msg = _check_wpad()
|
|
rep.add("WPAD / PAC", wpad_ok, wpad_msg)
|
|
|
|
# Per-User flag
|
|
pu_ok, pu_msg = _check_per_user_flag()
|
|
rep.add("ProxySettingsPerUser", pu_ok, pu_msg,
|
|
"needs Admin fix on this box" if not pu_ok else "")
|
|
|
|
# Policy locks
|
|
pol = detect_policy_overrides()
|
|
if pol:
|
|
rep.add("Group Policy proxy locks", False, f"{len(pol)} entries",
|
|
"policy keys override our proxy")
|
|
else:
|
|
rep.add("Group Policy proxy locks", True, "none")
|
|
|
|
# LAN-scope broadcasts
|
|
lan = lan_status()
|
|
lan_clean = (
|
|
sys.platform != "win32"
|
|
or ("OFF" in lan["llmnr"] and "OFF" in lan["mdns"] and "NICs with NetBIOS disabled" in lan["netbios"])
|
|
)
|
|
rep.add(
|
|
"LAN broadcast (LLMNR/NetBIOS/mDNS)",
|
|
lan_clean,
|
|
f"LLMNR={lan['llmnr']} NetBIOS={lan['netbios']} mDNS={lan['mdns']}",
|
|
"hostname leaks to local network" if not lan_clean else "",
|
|
)
|
|
|
|
# VPN
|
|
vpn = detect_vpn()
|
|
rep.add(
|
|
"VPN adapter",
|
|
True,
|
|
f"{vpn.label}" + (f" ({vpn.adapter})" if vpn.adapter else ""),
|
|
"informational",
|
|
)
|
|
|
|
# WebRTC policy
|
|
rtc_ok, rtc_msg = _check_webrtc_policy()
|
|
rep.add("Browser WebRTC protection", rtc_ok, rtc_msg)
|
|
|
|
# Admin status (a lot of fixes require it)
|
|
rep.add(
|
|
"Privilege model",
|
|
True if sys.platform == "darwin" else is_admin(),
|
|
"macOS prompts when network changes need approval" if sys.platform == "darwin" else ("Yes" if is_admin() else "No"),
|
|
"" if sys.platform == "darwin" or is_admin() else "MAC/IPv6/LAN/HKLM fixes need elevation",
|
|
)
|
|
|
|
return rep
|
|
|
|
|
|
def run_audit_sync(
|
|
listen_proxy: str,
|
|
ip_check_url: str,
|
|
timeout_seconds: float = 10.0,
|
|
) -> AuditReport:
|
|
return asyncio.run(run_audit(listen_proxy, ip_check_url, timeout_seconds))
|