From 6561bdc37fc5cce31e6b46f1632afed68864aa54 Mon Sep 17 00:00:00 2001 From: Indiana Holmes Date: Sat, 16 May 2026 18:40:53 -0700 Subject: [PATCH] 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 --- proxy_chain_manager/app.py | 77 ++++++++ proxy_chain_manager/config.py | 2 + proxy_chain_manager/leak_audit.py | 298 +++++++++++++++++++++++++++++ proxy_chain_manager/privacy_lan.py | 291 ++++++++++++++++++++++++++++ proxy_chain_manager/service.py | 29 +++ tests/test_proxy_god.py | 59 ++++++ 6 files changed, 756 insertions(+) create mode 100644 proxy_chain_manager/leak_audit.py create mode 100644 proxy_chain_manager/privacy_lan.py diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index 456bcba..37e4a29 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -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()), diff --git a/proxy_chain_manager/config.py b/proxy_chain_manager/config.py index 592c6cf..74b3517 100644 --- a/proxy_chain_manager/config.py +++ b/proxy_chain_manager/config.py @@ -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 = "" diff --git a/proxy_chain_manager/leak_audit.py b/proxy_chain_manager/leak_audit.py new file mode 100644 index 0000000..4841238 --- /dev/null +++ b/proxy_chain_manager/leak_audit.py @@ -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)) diff --git a/proxy_chain_manager/privacy_lan.py b/proxy_chain_manager/privacy_lan.py new file mode 100644 index 0000000..734c908 --- /dev/null +++ b/proxy_chain_manager/privacy_lan.py @@ -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 diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index 10a534c..07f58b0 100644 --- a/proxy_chain_manager/service.py +++ b/proxy_chain_manager/service.py @@ -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({ diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py index e915e3b..34c9f79 100644 --- a/tests/test_proxy_god.py +++ b/tests/test_proxy_god.py @@ -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()