Add LAN-broadcast lockdown, per-rotation MAC, leak audit panel
Tier-1 paranoid hardening: privacy_lan.py disables LLMNR/NetBIOS/mDNS with reversible snapshot; service rotates MAC on every chain rotation when enabled; leak_audit.py probes every leak surface (IP, DNS, IPv6, WPAD, GPO, ProxySettingsPerUser, LAN broadcast, VPN, WebRTC) and renders pass/fail in Privacy tab. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Indiana Holmes
@@ -36,6 +36,7 @@ from .firewall import (
|
||||
request_admin_relaunch,
|
||||
)
|
||||
from .dns_leak import check_dns_leak_hint, flush_dns_cache
|
||||
from .leak_audit import AuditReport, run_audit_sync
|
||||
from .browser_launcher import (
|
||||
BrowserConfig,
|
||||
BrowserSession,
|
||||
@@ -1083,10 +1084,12 @@ def main() -> None:
|
||||
"Applied on Start, restored on Stop. Admin required for MAC, hostname, IPv6, and WebRTC.",
|
||||
)
|
||||
mac_var = ctk.BooleanVar(value=s.mac_spoof_enabled)
|
||||
mac_rotate_var = ctk.BooleanVar(value=s.mac_rotate_on_chain_rotate)
|
||||
host_var = ctk.BooleanVar(value=s.spoof_hostname_enabled)
|
||||
dns_flush_var = ctk.BooleanVar(value=s.flush_dns_on_rotate)
|
||||
ipv6_var = ctk.BooleanVar(value=s.disable_ipv6_while_active)
|
||||
webrtc_var = ctk.BooleanVar(value=s.harden_webrtc_enabled)
|
||||
lan_var = ctk.BooleanVar(value=s.lan_lockdown_enabled)
|
||||
|
||||
def _priv_toggle(parent: Any, text: str, var: ctk.BooleanVar, tip: str = "") -> ctk.CTkCheckBox:
|
||||
cb = ctk.CTkCheckBox(
|
||||
@@ -1100,6 +1103,10 @@ def main() -> None:
|
||||
toggles_card, "Randomize MAC addresses on physical adapters", mac_var,
|
||||
"Changes NIC MAC values while chain runs (admin required).",
|
||||
)
|
||||
_priv_toggle(
|
||||
toggles_card, "Re-randomize MAC on EVERY chain rotation (paranoid)", mac_rotate_var,
|
||||
"Mutates MAC every time the chain rotates — prevents long-session correlation.",
|
||||
)
|
||||
_priv_toggle(
|
||||
toggles_card, "Spoof computer / NetBIOS hostname", host_var,
|
||||
"Temporarily renames machine identity while active (admin required).",
|
||||
@@ -1118,6 +1125,13 @@ def main() -> None:
|
||||
webrtc_var,
|
||||
"Applies Windows policy to block direct WebRTC UDP bypass in Chromium browsers.",
|
||||
)
|
||||
_priv_toggle(
|
||||
toggles_card,
|
||||
"LAN lockdown: kill LLMNR / NetBIOS / mDNS hostname broadcasts",
|
||||
lan_var,
|
||||
"Stops your machine from advertising its hostname on the local network "
|
||||
"(admin required, reversible on stop).",
|
||||
)
|
||||
|
||||
fp_card = _priv_section(
|
||||
"Device fingerprint",
|
||||
@@ -1163,6 +1177,65 @@ def main() -> None:
|
||||
_btn(dns_btn_row, "Flush DNS now", _manual_dns_flush, w=110, h=28,
|
||||
fg_color=DIM).pack(side="left")
|
||||
|
||||
audit_card = _priv_section(
|
||||
"Leak audit (mission-critical)",
|
||||
"Probes every leak surface: IP, DNS, IPv6, WPAD, Group Policy, "
|
||||
"ProxySettingsPerUser, LLMNR / NetBIOS / mDNS, VPN, WebRTC policy.",
|
||||
)
|
||||
audit_box = ctk.CTkTextbox(
|
||||
audit_card, height=220, font=("Consolas", 10),
|
||||
fg_color=BG, text_color=TEXT, scrollbar_button_color=ACCENT,
|
||||
)
|
||||
audit_box.pack(fill="x", padx=12, pady=(4, 4))
|
||||
audit_box.insert("end", "Click 'Run audit' to probe live leak status.")
|
||||
audit_status_lbl = ctk.CTkLabel(
|
||||
audit_card, text="", font=(FONT, 11, "bold"), text_color=TEXT2,
|
||||
)
|
||||
audit_status_lbl.pack(anchor="w", padx=12, pady=(0, 4))
|
||||
|
||||
def _render_audit(rep: AuditReport) -> None:
|
||||
audit_box.delete("1.0", "end")
|
||||
for f in rep.findings:
|
||||
mark = "OK " if f.ok else "LEAK"
|
||||
line = f"[{mark}] {f.name:<34} {f.value}"
|
||||
if f.note:
|
||||
line += f" — {f.note}"
|
||||
audit_box.insert("end", line + "\n")
|
||||
bad = sum(1 for f in rep.findings if not f.ok)
|
||||
if rep.overall_ok:
|
||||
audit_status_lbl.configure(
|
||||
text="CLEAN — no leaks detected on probed surfaces.", text_color=GREEN
|
||||
)
|
||||
else:
|
||||
audit_status_lbl.configure(
|
||||
text=f"{bad} leak(s) detected — see report above.", text_color=RED
|
||||
)
|
||||
|
||||
def _run_audit_async() -> None:
|
||||
audit_status_lbl.configure(text="Running audit…", text_color=YELLOW)
|
||||
audit_box.delete("1.0", "end")
|
||||
audit_box.insert("end", "Probing every leak surface in parallel…\n")
|
||||
|
||||
def work() -> None:
|
||||
try:
|
||||
rep = run_audit_sync(
|
||||
f"http://{svc.settings.listen_addr()}",
|
||||
svc.settings.ip_check_url,
|
||||
timeout_seconds=min(12.0, svc.settings.validation_timeout_seconds),
|
||||
)
|
||||
except Exception as e:
|
||||
root.after(0, lambda: audit_status_lbl.configure(
|
||||
text=f"Audit failed: {e}", text_color=RED))
|
||||
return
|
||||
root.after(0, lambda: _render_audit(rep))
|
||||
|
||||
threading.Thread(target=work, daemon=True).start()
|
||||
|
||||
audit_btn_row = ctk.CTkFrame(audit_card, fg_color="transparent")
|
||||
audit_btn_row.pack(fill="x", padx=12, pady=(0, 10))
|
||||
_btn(audit_btn_row, "Run audit", _run_audit_async, w=120, h=28,
|
||||
fg_color=ACCENT2, hover_color=ACCENT).pack(side="left", padx=(0, 8))
|
||||
|
||||
_btn(priv_scroll, "Save privacy settings", lambda: _save_settings(verbose=True),
|
||||
w=200, h=34, font=(FONT, 12)).pack(anchor="w", padx=8, pady=12)
|
||||
|
||||
@@ -1232,9 +1305,11 @@ def main() -> None:
|
||||
ks_var.set(True)
|
||||
dns_flush_var.set(True)
|
||||
mac_var.set(False)
|
||||
mac_rotate_var.set(False)
|
||||
host_var.set(False)
|
||||
ipv6_var.set(False)
|
||||
webrtc_var.set(True)
|
||||
lan_var.set(False)
|
||||
br_force_proxy_var.set(True)
|
||||
br_disable_webrtc_var.set(True)
|
||||
br_rfp_var.set(True)
|
||||
@@ -1300,10 +1375,12 @@ def main() -> None:
|
||||
sources=src_list or Settings().sources,
|
||||
ip_check_url=entries["check_url"].get().strip() or Settings().ip_check_url,
|
||||
mac_spoof_enabled=bool(mac_var.get()),
|
||||
mac_rotate_on_chain_rotate=bool(mac_rotate_var.get()),
|
||||
spoof_hostname_enabled=bool(host_var.get()),
|
||||
flush_dns_on_rotate=bool(dns_flush_var.get()),
|
||||
disable_ipv6_while_active=bool(ipv6_var.get()),
|
||||
harden_webrtc_enabled=bool(webrtc_var.get()),
|
||||
lan_lockdown_enabled=bool(lan_var.get()),
|
||||
firefox_path=firefox_path_var.get().strip(),
|
||||
firefox_profile_dir=firefox_profile_var.get().strip(),
|
||||
browser_clear_on_close=bool(br_clear_var.get()),
|
||||
|
||||
@@ -188,10 +188,12 @@ class Settings:
|
||||
|
||||
# ── privacy / device hardening (Privacy tab) ───────────────────────────
|
||||
mac_spoof_enabled: bool = False # randomize NIC MAC while chain runs (Admin)
|
||||
mac_rotate_on_chain_rotate: bool = False # re-randomize MAC on every chain rotation
|
||||
spoof_hostname_enabled: bool = False # temporary computer name while chain runs (Admin)
|
||||
flush_dns_on_rotate: bool = True # ipconfig /flushdns on each rotation
|
||||
disable_ipv6_while_active: bool = False # disable IPv6 bindings while chain runs (Admin)
|
||||
harden_webrtc_enabled: bool = False # Chrome/Edge WebRTC policy (Admin)
|
||||
lan_lockdown_enabled: bool = False # kill LLMNR/NetBIOS/mDNS while chain runs
|
||||
|
||||
# ── hardened browser ────────────────────────────────────────────────────
|
||||
firefox_path: str = ""
|
||||
|
||||
298
proxy_chain_manager/leak_audit.py
Normal file
298
proxy_chain_manager/leak_audit.py
Normal file
@@ -0,0 +1,298 @@
|
||||
"""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 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).
|
||||
"""
|
||||
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]:
|
||||
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]:
|
||||
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)."""
|
||||
paths = (
|
||||
r"SOFTWARE\Policies\Google\Chrome",
|
||||
r"SOFTWARE\Policies\Microsoft\Edge",
|
||||
)
|
||||
found = []
|
||||
for p in paths:
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, p, 0, winreg.KEY_QUERY_VALUE
|
||||
) as key:
|
||||
v = int(winreg.QueryValueEx(key, "DefaultWebRtcIpHandlingPolicy")[0])
|
||||
if v == 2:
|
||||
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 = "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 policy", rtc_ok, rtc_msg)
|
||||
|
||||
# Admin status (a lot of fixes require it)
|
||||
rep.add(
|
||||
"Administrator privileges",
|
||||
is_admin(),
|
||||
"Yes" if is_admin() else "No",
|
||||
"MAC/IPv6/LAN/HKLM fixes need elevation" if not is_admin() else "",
|
||||
)
|
||||
|
||||
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))
|
||||
291
proxy_chain_manager/privacy_lan.py
Normal file
291
proxy_chain_manager/privacy_lan.py
Normal file
@@ -0,0 +1,291 @@
|
||||
"""LAN-scope hostname leak lockdown.
|
||||
|
||||
Windows by default broadcasts the local hostname on every adapter via:
|
||||
- LLMNR (UDP 5355) — link-local multicast name resolution
|
||||
- NetBIOS over TCP/IP — broadcast name registration on the subnet
|
||||
- mDNS (UDP 5353) — Bonjour-style name advertising on newer Windows
|
||||
|
||||
Anyone on the same Wi-Fi / VLAN can grab the hostname and tie it to the MAC
|
||||
address. Killing these is a paranoid-mode no-brainer.
|
||||
|
||||
All three are reversible. We snapshot original state on engage and restore on
|
||||
disengage.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import subprocess
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import winreg
|
||||
|
||||
from .firewall import is_admin
|
||||
from .win_compat import probe as _win_probe
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_LLMNR_POLICY_PATH = r"SOFTWARE\Policies\Microsoft\Windows NT\DNSClient"
|
||||
_LLMNR_VALUE = "EnableMulticast"
|
||||
_MDNS_DNSCLIENT = r"SYSTEM\CurrentControlSet\Services\Dnscache\Parameters"
|
||||
_MDNS_VALUE = "EnableMDNS"
|
||||
|
||||
|
||||
@dataclass
|
||||
class LanSnapshot:
|
||||
netbios_per_interface: dict[str, int] = field(default_factory=dict)
|
||||
llmnr_present: bool = False
|
||||
llmnr_prev_value: int | None = None
|
||||
mdns_present: bool = False
|
||||
mdns_prev_value: int | None = None
|
||||
|
||||
|
||||
def _run(args: list[str], timeout: float = 12.0) -> tuple[int, str, str]:
|
||||
try:
|
||||
r = subprocess.run(
|
||||
args,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||||
)
|
||||
return r.returncode, r.stdout or "", r.stderr or ""
|
||||
except Exception as e:
|
||||
return 1, "", str(e)
|
||||
|
||||
|
||||
def _list_netbios_via_wmic() -> dict[str, int]:
|
||||
"""Per-NIC NetBIOS setting via wmic (works back to Server 2008 R2).
|
||||
|
||||
Returns: { settings_id : tcpip_netbios_options }
|
||||
0 = use DHCP, 1 = enabled, 2 = disabled
|
||||
"""
|
||||
code, out, _ = _run([
|
||||
"wmic", "nicconfig", "where", "IPEnabled=true",
|
||||
"get", "SettingID,TcpipNetbiosOptions", "/format:list",
|
||||
])
|
||||
if code != 0 or not out:
|
||||
return {}
|
||||
settings: dict[str, int] = {}
|
||||
cur_id = ""
|
||||
cur_val: int | None = None
|
||||
for raw in out.splitlines():
|
||||
line = raw.strip()
|
||||
if not line:
|
||||
if cur_id and cur_val is not None:
|
||||
settings[cur_id] = cur_val
|
||||
cur_id, cur_val = "", None
|
||||
continue
|
||||
if line.startswith("SettingID="):
|
||||
cur_id = line.split("=", 1)[1].strip()
|
||||
elif line.startswith("TcpipNetbiosOptions="):
|
||||
try:
|
||||
cur_val = int(line.split("=", 1)[1].strip())
|
||||
except ValueError:
|
||||
cur_val = None
|
||||
if cur_id and cur_val is not None:
|
||||
settings[cur_id] = cur_val
|
||||
return settings
|
||||
|
||||
|
||||
def _set_netbios_via_wmic(settings_id: str, mode: int) -> bool:
|
||||
"""mode: 0=DHCP, 1=enabled, 2=disabled."""
|
||||
code, _, _ = _run([
|
||||
"wmic", "nicconfig", "where",
|
||||
f"SettingID='{settings_id}'",
|
||||
"call", "SetTcpipNetbios", str(mode),
|
||||
])
|
||||
return code == 0
|
||||
|
||||
|
||||
def _llmnr_read() -> tuple[bool, int | None]:
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, _LLMNR_POLICY_PATH, 0, winreg.KEY_QUERY_VALUE
|
||||
) as key:
|
||||
val, _ = winreg.QueryValueEx(key, _LLMNR_VALUE)
|
||||
return True, int(val)
|
||||
except OSError:
|
||||
return False, None
|
||||
|
||||
|
||||
def _llmnr_write(value: int) -> bool:
|
||||
try:
|
||||
with winreg.CreateKeyEx(
|
||||
winreg.HKEY_LOCAL_MACHINE, _LLMNR_POLICY_PATH, 0, winreg.KEY_SET_VALUE
|
||||
) as key:
|
||||
winreg.SetValueEx(key, _LLMNR_VALUE, 0, winreg.REG_DWORD, value)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _llmnr_delete() -> None:
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, _LLMNR_POLICY_PATH, 0, winreg.KEY_SET_VALUE
|
||||
) as key:
|
||||
winreg.DeleteValue(key, _LLMNR_VALUE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _mdns_read() -> tuple[bool, int | None]:
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, _MDNS_DNSCLIENT, 0, winreg.KEY_QUERY_VALUE
|
||||
) as key:
|
||||
val, _ = winreg.QueryValueEx(key, _MDNS_VALUE)
|
||||
return True, int(val)
|
||||
except OSError:
|
||||
return False, None
|
||||
|
||||
|
||||
def _mdns_write(value: int) -> bool:
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, _MDNS_DNSCLIENT, 0, winreg.KEY_SET_VALUE
|
||||
) as key:
|
||||
winreg.SetValueEx(key, _MDNS_VALUE, 0, winreg.REG_DWORD, value)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _restart_dnscache() -> None:
|
||||
"""Bounce the DNS Client service so the LLMNR/mDNS toggles take effect.
|
||||
|
||||
Server hardening baselines sometimes mark this service "denied" — failure
|
||||
is logged and ignored; the change still applies on next reboot.
|
||||
"""
|
||||
_run(["net", "stop", "Dnscache", "/y"], timeout=20)
|
||||
_run(["net", "start", "Dnscache"], timeout=20)
|
||||
|
||||
|
||||
def lan_status() -> dict[str, str]:
|
||||
"""Human-readable current state. Returned even when not Admin."""
|
||||
nb = _list_netbios_via_wmic()
|
||||
if nb:
|
||||
disabled = sum(1 for v in nb.values() if v == 2)
|
||||
nb_summary = f"{disabled}/{len(nb)} NICs with NetBIOS disabled"
|
||||
else:
|
||||
nb_summary = "unavailable"
|
||||
llmnr_present, llmnr_v = _llmnr_read()
|
||||
if not llmnr_present:
|
||||
llmnr_summary = "default (ON)"
|
||||
else:
|
||||
llmnr_summary = "OFF" if llmnr_v == 0 else f"ON (policy={llmnr_v})"
|
||||
mdns_present, mdns_v = _mdns_read()
|
||||
if not mdns_present:
|
||||
mdns_summary = "default (ON on Win10 1903+)"
|
||||
else:
|
||||
mdns_summary = "OFF" if mdns_v == 0 else f"ON (EnableMDNS={mdns_v})"
|
||||
return {
|
||||
"netbios": nb_summary,
|
||||
"llmnr": llmnr_summary,
|
||||
"mdns": mdns_summary,
|
||||
}
|
||||
|
||||
|
||||
def engage_lan_lockdown() -> tuple[LanSnapshot | None, list[str]]:
|
||||
"""Disable LLMNR / NetBIOS / mDNS. Returns (snapshot for restore, log lines).
|
||||
|
||||
snapshot is None when the operation was refused (not Admin, etc.).
|
||||
"""
|
||||
logs: list[str] = []
|
||||
if not is_admin():
|
||||
return None, ["LAN privacy lockdown skipped — needs Administrator."]
|
||||
|
||||
snap = LanSnapshot()
|
||||
snap.netbios_per_interface = _list_netbios_via_wmic()
|
||||
snap.llmnr_present, snap.llmnr_prev_value = _llmnr_read()
|
||||
snap.mdns_present, snap.mdns_prev_value = _mdns_read()
|
||||
|
||||
if snap.netbios_per_interface:
|
||||
changed = 0
|
||||
for sid in snap.netbios_per_interface:
|
||||
if _set_netbios_via_wmic(sid, 2):
|
||||
changed += 1
|
||||
logs.append(f"NetBIOS over TCP/IP disabled on {changed}/{len(snap.netbios_per_interface)} NICs.")
|
||||
else:
|
||||
logs.append("NetBIOS: could not enumerate adapters (wmic missing?).")
|
||||
|
||||
if _llmnr_write(0):
|
||||
logs.append("LLMNR (UDP 5355) disabled via DNSClient policy.")
|
||||
else:
|
||||
logs.append("LLMNR disable failed (policy write rejected).")
|
||||
|
||||
if _mdns_write(0):
|
||||
logs.append("mDNS (UDP 5353) disabled via Dnscache parameters.")
|
||||
else:
|
||||
logs.append("mDNS disable failed (Dnscache parameters not writable).")
|
||||
|
||||
_restart_dnscache()
|
||||
logs.append("DNS Client service bounced — LAN lockdown active.")
|
||||
return snap, logs
|
||||
|
||||
|
||||
def restore_lan(snap: LanSnapshot | None) -> list[str]:
|
||||
"""Roll back to the snapshot captured by engage_lan_lockdown."""
|
||||
if snap is None or not is_admin():
|
||||
return []
|
||||
logs: list[str] = []
|
||||
|
||||
if snap.netbios_per_interface:
|
||||
restored = 0
|
||||
for sid, prev in snap.netbios_per_interface.items():
|
||||
if _set_netbios_via_wmic(sid, prev):
|
||||
restored += 1
|
||||
logs.append(f"NetBIOS restored on {restored}/{len(snap.netbios_per_interface)} NICs.")
|
||||
|
||||
if snap.llmnr_present and snap.llmnr_prev_value is not None:
|
||||
_llmnr_write(snap.llmnr_prev_value)
|
||||
logs.append(f"LLMNR policy restored to {snap.llmnr_prev_value}.")
|
||||
else:
|
||||
_llmnr_delete()
|
||||
logs.append("LLMNR policy removed (was default).")
|
||||
|
||||
if snap.mdns_present and snap.mdns_prev_value is not None:
|
||||
_mdns_write(snap.mdns_prev_value)
|
||||
logs.append(f"mDNS restored (EnableMDNS={snap.mdns_prev_value}).")
|
||||
else:
|
||||
# If mDNS key was absent originally, delete the value we created.
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE, _MDNS_DNSCLIENT, 0, winreg.KEY_SET_VALUE
|
||||
) as key:
|
||||
winreg.DeleteValue(key, _MDNS_VALUE)
|
||||
logs.append("mDNS toggle removed (was default).")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
_restart_dnscache()
|
||||
return logs
|
||||
|
||||
|
||||
def snapshot_to_json(snap: LanSnapshot | None) -> str:
|
||||
if snap is None:
|
||||
return ""
|
||||
return json.dumps({
|
||||
"netbios": snap.netbios_per_interface,
|
||||
"llmnr_present": snap.llmnr_present,
|
||||
"llmnr_prev_value": snap.llmnr_prev_value,
|
||||
"mdns_present": snap.mdns_present,
|
||||
"mdns_prev_value": snap.mdns_prev_value,
|
||||
})
|
||||
|
||||
|
||||
def snapshot_from_json(raw: str) -> LanSnapshot | None:
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
d = json.loads(raw)
|
||||
return LanSnapshot(
|
||||
netbios_per_interface={k: int(v) for k, v in (d.get("netbios") or {}).items()},
|
||||
llmnr_present=bool(d.get("llmnr_present")),
|
||||
llmnr_prev_value=d.get("llmnr_prev_value"),
|
||||
mdns_present=bool(d.get("mdns_present")),
|
||||
mdns_prev_value=d.get("mdns_prev_value"),
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
@@ -30,6 +30,11 @@ from .firewall import disengage as fw_disengage, engage as fw_engage, is_admin
|
||||
from .gost_util import build_gost_cmd, ensure_gost, popen_no_window, read_gost_log_tail, terminate_process
|
||||
from .leak_detect import is_chain_leak, leak_reason
|
||||
from .mac_spoof import restore_macs, spoof_all_physical
|
||||
from .privacy_lan import (
|
||||
LanSnapshot,
|
||||
engage_lan_lockdown,
|
||||
restore_lan,
|
||||
)
|
||||
from .sysproxy import (
|
||||
clear_system_proxy,
|
||||
detect_policy_overrides,
|
||||
@@ -76,6 +81,7 @@ class ChainService:
|
||||
self._hostname_original: str | None = None
|
||||
self._ipv6_adapters: list[str] = []
|
||||
self._webrtc_was_applied: bool = False
|
||||
self._lan_snap: LanSnapshot | None = None
|
||||
|
||||
def _manual_exit_url(self) -> str | None:
|
||||
u = normalize_proxy_url(self._settings.manual_exit_proxy)
|
||||
@@ -167,6 +173,14 @@ class ChainService:
|
||||
elif s.harden_webrtc_enabled:
|
||||
self._notify({"type": "log", "text": "WebRTC hardening enabled but not Admin — skipped."})
|
||||
|
||||
if s.lan_lockdown_enabled and is_admin():
|
||||
snap, logs = engage_lan_lockdown()
|
||||
self._lan_snap = snap
|
||||
for ln in logs:
|
||||
self._notify({"type": "log", "text": f"LAN: {ln}"})
|
||||
elif s.lan_lockdown_enabled:
|
||||
self._notify({"type": "log", "text": "LAN lockdown enabled but not Admin — skipped."})
|
||||
|
||||
def _restore_privacy(self) -> None:
|
||||
if self._mac_originals:
|
||||
for ln in restore_macs(self._mac_originals):
|
||||
@@ -184,6 +198,10 @@ class ChainService:
|
||||
apply_webrtc_hardening(False)
|
||||
self._webrtc_was_applied = False
|
||||
self._notify({"type": "log", "text": "WebRTC policy restored."})
|
||||
if self._lan_snap is not None and is_admin():
|
||||
for ln in restore_lan(self._lan_snap):
|
||||
self._notify({"type": "log", "text": f"LAN: {ln}"})
|
||||
self._lan_snap = None
|
||||
|
||||
def _run_thread(self) -> None:
|
||||
try:
|
||||
@@ -390,6 +408,17 @@ class ChainService:
|
||||
ok, msg = flush_dns_cache()
|
||||
if ok:
|
||||
log.debug("DNS cache flushed before chain run")
|
||||
# Per-rotation MAC re-randomization (only if mac_spoof is also on, since
|
||||
# without spoof there's no original snapshot we own to mutate).
|
||||
if (
|
||||
self._settings.mac_rotate_on_chain_rotate
|
||||
and self._settings.mac_spoof_enabled
|
||||
and is_admin()
|
||||
and self._mac_originals
|
||||
):
|
||||
_, logs = spoof_all_physical(self._mac_originals)
|
||||
for ln in logs:
|
||||
self._notify({"type": "log", "text": f"MAC rotate: {ln}"})
|
||||
listen = self._settings.listen_addr()
|
||||
cmd = build_gost_cmd(gost, listen, chain)
|
||||
self._notify({
|
||||
|
||||
@@ -217,5 +217,64 @@ class TestValidator(unittest.TestCase):
|
||||
self.assertIsNone(asyncio.run(run()))
|
||||
|
||||
|
||||
class TestPrivacyLan(unittest.TestCase):
|
||||
def test_snapshot_round_trip(self) -> None:
|
||||
from proxy_chain_manager.privacy_lan import (
|
||||
LanSnapshot, snapshot_from_json, snapshot_to_json,
|
||||
)
|
||||
s = LanSnapshot(
|
||||
netbios_per_interface={"{a-b-c}": 0, "{d-e-f}": 2},
|
||||
llmnr_present=True,
|
||||
llmnr_prev_value=1,
|
||||
mdns_present=False,
|
||||
mdns_prev_value=None,
|
||||
)
|
||||
round_tripped = snapshot_from_json(snapshot_to_json(s))
|
||||
self.assertIsNotNone(round_tripped)
|
||||
assert round_tripped is not None
|
||||
self.assertEqual(round_tripped.netbios_per_interface, s.netbios_per_interface)
|
||||
self.assertEqual(round_tripped.llmnr_present, s.llmnr_present)
|
||||
self.assertEqual(round_tripped.llmnr_prev_value, s.llmnr_prev_value)
|
||||
self.assertEqual(round_tripped.mdns_present, s.mdns_present)
|
||||
self.assertEqual(round_tripped.mdns_prev_value, s.mdns_prev_value)
|
||||
|
||||
def test_lan_status_returns_keys(self) -> None:
|
||||
from proxy_chain_manager.privacy_lan import lan_status
|
||||
st = lan_status()
|
||||
self.assertIn("netbios", st)
|
||||
self.assertIn("llmnr", st)
|
||||
self.assertIn("mdns", st)
|
||||
|
||||
|
||||
class TestLeakAuditPure(unittest.TestCase):
|
||||
def test_categorize_exit_private(self) -> None:
|
||||
from proxy_chain_manager.leak_audit import _categorize_exit
|
||||
self.assertEqual(_categorize_exit("10.0.0.5"), "PRIVATE — chain broken")
|
||||
self.assertEqual(_categorize_exit("192.168.1.1"), "PRIVATE — chain broken")
|
||||
self.assertEqual(_categorize_exit("172.20.5.5"), "PRIVATE — chain broken")
|
||||
|
||||
def test_categorize_exit_loopback(self) -> None:
|
||||
from proxy_chain_manager.leak_audit import _categorize_exit
|
||||
self.assertEqual(_categorize_exit("127.0.0.1"), "LOOPBACK — chain broken")
|
||||
|
||||
def test_categorize_exit_public(self) -> None:
|
||||
from proxy_chain_manager.leak_audit import _categorize_exit
|
||||
self.assertEqual(_categorize_exit("104.28.3.99"), "public")
|
||||
self.assertEqual(_categorize_exit("8.8.8.8"), "public")
|
||||
|
||||
def test_categorize_exit_none(self) -> None:
|
||||
from proxy_chain_manager.leak_audit import _categorize_exit
|
||||
self.assertEqual(_categorize_exit(None), "unknown")
|
||||
|
||||
def test_audit_finding_aggregation(self) -> None:
|
||||
from proxy_chain_manager.leak_audit import AuditReport
|
||||
rep = AuditReport()
|
||||
rep.add("X", True, "ok")
|
||||
self.assertTrue(rep.overall_ok)
|
||||
rep.add("Y", False, "bad", "note")
|
||||
self.assertFalse(rep.overall_ok)
|
||||
self.assertEqual(len(rep.findings), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user