From 8f012402a691a812da236233032037e8d8bb4f5b Mon Sep 17 00:00:00 2001 From: Indiana Holmes Date: Sat, 16 May 2026 18:03:30 -0700 Subject: [PATCH] Harden Windows Server compatibility and system-wide proxy. Fix exit-IP checks for HTTP-only proxies, apply proxy via WinINet Connections blob and WinHTTP, improve VPN detection on legacy Server, add SOCKS5 host:port:user:pass exit parsing, and add win_compat probe for PowerShell 2.0 hosts. Co-authored-by: Cursor --- proxy_chain_manager/app.py | 62 +++++++--- proxy_chain_manager/config.py | 51 +++++++++ proxy_chain_manager/fingerprint.py | 8 ++ proxy_chain_manager/mac_spoof.py | 49 +++++++- proxy_chain_manager/service.py | 32 +++++- proxy_chain_manager/sysproxy.py | 177 +++++++++++++++++++++++++++-- proxy_chain_manager/validator.py | 31 ++++- proxy_chain_manager/vpn_detect.py | 82 +++++++++++-- proxy_chain_manager/win_compat.py | 90 +++++++++++++++ tests/test_proxy_god.py | 40 +++++++ 10 files changed, 579 insertions(+), 43 deletions(-) create mode 100644 proxy_chain_manager/win_compat.py diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index d38ec57..456bcba 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -24,6 +24,7 @@ from .config import ( load_settings, merge_proxy_credentials, normalize_proxy_url, + parse_exit_proxy_input, redact_proxy_url, save_settings, split_proxy_for_edit, @@ -490,7 +491,7 @@ def main() -> None: exit_test_dot.pack(side="left", padx=(8, 4)) exit_proxy_entry = ctk.CTkEntry( exit_fix_frame, - placeholder_text="http://host:port or socks5://host:port (optional)", + placeholder_text="host:port:user:pass (defaults to socks5 — paste any format)", font=("Consolas", 10), fg_color=BG, border_color=ACCENT, @@ -527,13 +528,49 @@ def main() -> None: exit_btn_row = ctk.CTkFrame(exit_fix_frame, fg_color="transparent") exit_btn_row.pack(fill="x", padx=10, pady=(0, 4)) + def _auto_split_exit_paste(_event: Any | None = None) -> None: + """If the URL field looks like ``host:port:user:pass``, distribute + the user/pass parts into their own fields so the user sees what got + parsed. Triggered on key release and explicit paste. + """ + raw = exit_proxy_entry.get() + if "://" in raw: + return + bits = [b for b in raw.strip().split(":") if b != ""] + if len(bits) < 4 and "@" not in raw and not any(c in raw for c in (" ", "\t")): + return + base, user, pw = parse_exit_proxy_input(raw, default_scheme="socks5") + if not base: + return + if base != raw: + exit_proxy_entry.delete(0, "end") + exit_proxy_entry.insert(0, base) + if user and not exit_user_entry.get().strip(): + exit_user_entry.delete(0, "end") + exit_user_entry.insert(0, user) + if pw and not exit_pass_entry.get().strip(): + exit_pass_entry.delete(0, "end") + exit_pass_entry.insert(0, pw) + + exit_proxy_entry.bind("", _auto_split_exit_paste) + exit_proxy_entry.bind("<>", lambda e: exit_proxy_entry.after(1, _auto_split_exit_paste)) + exit_proxy_entry.bind("", _auto_split_exit_paste) + def _exit_url() -> str: - raw = exit_proxy_entry.get().strip() - if not raw: + """Build the final exit-proxy URL. + + Accepts every paste format ``parse_exit_proxy_input`` understands + (host:port:user:pass, user:pass@host:port, scheme://…). Defaults to + SOCKS5 when no scheme is supplied. + """ + base, parsed_user, parsed_pw = parse_exit_proxy_input( + exit_proxy_entry.get(), default_scheme="socks5" + ) + if not base: return "" - if "://" not in raw: - raw = "http://" + raw - return merge_proxy_credentials(raw, exit_user_entry.get(), exit_pass_entry.get()) + user = (exit_user_entry.get() or "").strip() or parsed_user + pw = (exit_pass_entry.get() or "").strip() or parsed_pw + return merge_proxy_credentials(base, user, pw) def _test_exit() -> None: url = _exit_url() @@ -567,8 +604,11 @@ def main() -> None: ctk.CTkLabel( exit_fix_frame, - text="Appended after your chain hops when running. Leave empty to use last chain hop as exit.\n" - "Auth: User/Pass fields or user:pass@host in the URL.", + text=( + "Appended after your chain hops. Leave empty to use last chain hop as exit.\n" + "Paste any of: host:port:user:pass • user:pass@host:port • scheme://host:port\n" + "No scheme = SOCKS5. Paste auto-splits into User/Pass below." + ), font=(FONT, 9), text_color=TEXT2, justify="left", @@ -1247,11 +1287,7 @@ def main() -> None: obfuscation_mode=mode_var.get(), use_pinned_chain=bool(use_manual_var.get()), pinned_chain=list(manual_chain), - manual_exit_proxy=merge_proxy_credentials( - exit_proxy_entry.get(), - exit_user_entry.get(), - exit_pass_entry.get(), - ), + manual_exit_proxy=_exit_url(), health_check_seconds=min(3600, max(10, int(entries["health"].get().strip()))), full_refresh_seconds=min(86400, max(60, int(entries["refresh"].get().strip()))), validation_concurrency=max(1, int(entries["conc"].get().strip())), diff --git a/proxy_chain_manager/config.py b/proxy_chain_manager/config.py index 8f90f98..592c6cf 100644 --- a/proxy_chain_manager/config.py +++ b/proxy_chain_manager/config.py @@ -78,6 +78,57 @@ def merge_proxy_credentials(raw_url: str, username: str = "", password: str = "" return urlunparse((scheme, netloc, "", "", "", "")).rstrip("/") +def parse_exit_proxy_input( + raw: str, default_scheme: str = "socks5" +) -> tuple[str, str, str]: + """Parse any sane paste format into ``(base_url, user, password)``. + + Accepts (whitespace-trimmed, default scheme = SOCKS5 when none given): + + ``host:port`` + ``host:port:user:password`` + ``host:port:user:password:session_or_extra`` (extras folded into password) + ``host:port user password`` (tab/space-separated) + ``user:password@host:port`` + ``scheme://host:port`` + ``scheme://user:password@host:port`` + + User and password may contain ``@``, ``:`` and arbitrary "words" — those + are URL-encoded automatically when ``merge_proxy_credentials`` rebuilds + the URL, so pasting plain text is safe. + """ + t = (raw or "").strip() + if not t: + return "", "", "" + + if "://" in t: + # Already a full URL — defer to the existing splitter. + return split_proxy_for_edit(t) + + if "@" in t: + return split_proxy_for_edit(f"{default_scheme}://{t}") + + if "\t" in t or any(c in t for c in (" ",)): + # "host:port user pass" or "host:port\tuser\tpass" + parts = [p for p in t.replace("\t", " ").split() if p] + if len(parts) >= 3: + hp, user, pw = parts[0], parts[1], " ".join(parts[2:]) + return f"{default_scheme}://{hp}".rstrip("/"), user, pw + if len(parts) == 2: + return f"{default_scheme}://{parts[0]}".rstrip("/"), "", "" + t = parts[0] + + bits = t.split(":") + if len(bits) >= 4: + host, port, user = bits[0], bits[1], bits[2] + password = ":".join(bits[3:]) # passwords containing ":" survive + return f"{default_scheme}://{host}:{port}", user, password + if len(bits) == 2: + return f"{default_scheme}://{bits[0]}:{bits[1]}", "", "" + # 1 token, or 3-token oddity — best-effort; treat as host[:port] only. + return f"{default_scheme}://{t}".rstrip("/"), "", "" + + def redact_proxy_url(url: str) -> str: """Same URL shape for logs/UI, with credentials replaced by ``***``.""" t = (url or "").strip() diff --git a/proxy_chain_manager/fingerprint.py b/proxy_chain_manager/fingerprint.py index 99405b2..3b93bb5 100644 --- a/proxy_chain_manager/fingerprint.py +++ b/proxy_chain_manager/fingerprint.py @@ -11,6 +11,7 @@ from dataclasses import dataclass, field from .firewall import is_admin from .mac_spoof import list_nics +from .win_compat import probe as _win_probe log = logging.getLogger(__name__) @@ -106,6 +107,11 @@ def disable_ipv6_on_adapters() -> tuple[list[str], list[str]]: """Disable IPv6 binding on up physical adapters. Returns (adapter names, log lines).""" if not is_admin(): return [], ["IPv6 disable skipped (not Admin)."] + if not _win_probe().has_net_cmdlets: + return [], [ + "IPv6 disable skipped — requires PowerShell 3.0+ " + "(Windows 8 / Server 2012+). Detected legacy PowerShell." + ] script = ( "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " "ForEach-Object { $_.Name }" @@ -135,6 +141,8 @@ def disable_ipv6_on_adapters() -> tuple[list[str], list[str]]: def enable_ipv6_on_adapters(adapters: list[str]) -> list[str]: if not is_admin() or not adapters: return [] + if not _win_probe().has_net_cmdlets: + return [] logs: list[str] = [] for name in adapters: cmd = ( diff --git a/proxy_chain_manager/mac_spoof.py b/proxy_chain_manager/mac_spoof.py index d2f111d..e6d9872 100644 --- a/proxy_chain_manager/mac_spoof.py +++ b/proxy_chain_manager/mac_spoof.py @@ -8,6 +8,7 @@ import subprocess from dataclasses import dataclass from .firewall import is_admin +from .win_compat import probe as _win_probe log = logging.getLogger(__name__) @@ -36,7 +37,14 @@ def _run(args: list[str], timeout: float = 20.0) -> tuple[int, str, str]: def list_nics() -> list[NicMac]: - """Physical/up adapters with current MAC.""" + """Physical/up adapters with current MAC. + + Falls back to ``wmic nic`` for hosts without PowerShell 3.0 (Server 2008 R2). + """ + compat = _win_probe() + if not compat.has_net_cmdlets: + return _list_nics_wmic() + script = ( "Get-NetAdapter | Where-Object { $_.Status -ne 'Disabled' } | " "Select-Object Name, MacAddress, InterfaceDescription | " @@ -44,7 +52,7 @@ def list_nics() -> list[NicMac]: ) code, out, _ = _run(["powershell", "-NoProfile", "-Command", script]) if code != 0 or not out.strip(): - return [] + return _list_nics_wmic() import json try: @@ -71,9 +79,46 @@ def random_mac() -> str: return ":".join(f"{x:02X}" for x in b) +def _list_nics_wmic() -> list[NicMac]: + """Server 2008 R2 / PowerShell 2.0 fallback via wmic.""" + code, out, _ = _run([ + "wmic", "nic", "where", "NetEnabled=true", + "get", "NetConnectionID,MACAddress,Name", "/format:list", + ]) + if code != 0 or not out: + return [] + nics: list[NicMac] = [] + cur = {"id": "", "mac": "", "name": ""} + for ln in out.splitlines(): + s = ln.strip() + if not s: + if cur["id"] and cur["mac"]: + mac = cur["mac"].replace("-", ":") + if mac and mac != "00:00:00:00:00:00": + nics.append(NicMac(name=cur["id"], mac=mac, description=cur["name"])) + cur = {"id": "", "mac": "", "name": ""} + continue + if s.startswith("NetConnectionID="): + cur["id"] = s.split("=", 1)[1].strip() + elif s.startswith("MACAddress="): + cur["mac"] = s.split("=", 1)[1].strip() + elif s.startswith("Name="): + cur["name"] = s.split("=", 1)[1].strip() + if cur["id"] and cur["mac"]: + mac = cur["mac"].replace("-", ":") + if mac and mac != "00:00:00:00:00:00": + nics.append(NicMac(name=cur["id"], mac=mac, description=cur["name"])) + return nics + + def set_mac(adapter: str, mac: str) -> tuple[bool, str]: if not is_admin(): return False, "Administrator required to change MAC." + if not _win_probe().has_net_cmdlets: + return False, ( + "MAC change requires PowerShell 3.0+ (Windows 8 / Server 2012+). " + "Detected legacy PowerShell — feature unavailable on this host." + ) mac = mac.replace("-", ":").upper() if not _MAC_RE.match(mac.replace(":", "-")): return False, f"Invalid MAC: {mac}" diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index 6356a8a..bff8cee 100644 --- a/proxy_chain_manager/service.py +++ b/proxy_chain_manager/service.py @@ -30,9 +30,10 @@ 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 .sysproxy import clear_system_proxy, set_system_proxy +from .sysproxy import clear_system_proxy, is_system_proxy_set, set_system_proxy from .validator import check_chain_exit_ip, get_direct_ip, validate_proxies from .vpn_detect import VpnStatus, detect_vpn +from .win_compat import probe as _win_probe log = logging.getLogger(__name__) @@ -89,6 +90,17 @@ class ChainService: def start(self) -> None: if self._thread and self._thread.is_alive(): return + compat = _win_probe() + self._notify({"type": "log", "text": f"Host: {compat.summary()}"}) + if not compat.has_net_cmdlets: + self._notify({ + "type": "log", + "text": ( + "Legacy PowerShell detected — MAC spoof / IPv6 disable / " + "hostname spoof unavailable on this host. Proxy chain + " + "kill-switch + WebRTC policy still work." + ), + }) self._stop.clear() self._thread = threading.Thread( target=self._run_thread, name="ChainService", daemon=True @@ -419,7 +431,9 @@ class ChainService: "text": f"Exit IP check through local proxy (≤{int(timeout * 2 + 5)}s)…", }) t0 = time.monotonic() - exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout) + exit_ip = await check_chain_exit_ip( + local_proxy, self._settings.ip_check_url, timeout, chain_hops=len(chain) + ) log.debug("Initial exit IP check took %.2fs → %s", time.monotonic() - t0, exit_ip or "none") if not exit_ip: @@ -453,7 +467,15 @@ class ChainService: self._settings.local_port, self._settings.proxy_bypass, ) - self._notify({"type": "log", "text": f"System proxy → {self._settings.listen_addr()}"}) + applied = is_system_proxy_set() + self._notify({ + "type": "log", + "text": ( + f"System proxy → {self._settings.listen_addr()} " + f"(WinINet root + Connections + WinHTTP) " + f"{'OK' if applied else 'FAILED — registry write rejected'}" + ), + }) # ── Health monitor loop ─────────────────────────────────────────────── hc = int(self._settings.health_check_seconds) @@ -470,7 +492,9 @@ class ChainService: self._notify({"type": "log", "text": "Health check..."}) t1 = time.monotonic() - exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout) + exit_ip = await check_chain_exit_ip( + local_proxy, self._settings.ip_check_url, timeout, chain_hops=len(chain) + ) log.debug("Periodic exit IP check %.2fs → %s", time.monotonic() - t1, exit_ip or "none") if is_chain_leak(exit_ip, real_ip, self._vpn.active): diff --git a/proxy_chain_manager/sysproxy.py b/proxy_chain_manager/sysproxy.py index 5330be0..187fcec 100644 --- a/proxy_chain_manager/sysproxy.py +++ b/proxy_chain_manager/sysproxy.py @@ -1,16 +1,41 @@ -"""Set / clear the Windows system-wide HTTP proxy via the registry + WinINet broadcast.""" +"""Set / clear the Windows system-wide HTTP proxy. + +Three layers get configured so the proxy actually applies system-wide: + +1. ``HKCU\\…\\Internet Settings`` — simple values WinINet reads first +2. ``Internet Settings\\Connections\\DefaultConnectionSettings`` binary blob — + overrides #1 for any process that opens the per-connection structure. On + enterprise images and Windows Server boxes this stale blob is the usual + reason "chain says connected but the browser still leaks direct". +3. ``netsh winhttp set proxy`` — covers .NET, Windows Update, many + services and headless browser components that bypass WinINet. + +Conflicting overrides (``AutoConfigURL`` / ``AutoDetect`` WPAD) are wiped on +engage and restored to "off" on disengage. +""" from __future__ import annotations import ctypes import logging +import struct +import subprocess import winreg log = logging.getLogger(__name__) _KEY_PATH = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings" +_CONN_KEY_PATH = ( + r"Software\Microsoft\Windows\CurrentVersion\Internet Settings\Connections" +) _SETTINGS_CHANGED = 39 _REFRESH = 37 +# Flags inside the DefaultConnectionSettings binary blob +_FLAG_DIRECT = 0x01 +_FLAG_MANUAL_PROXY = 0x02 +_FLAG_AUTO_CONFIG = 0x04 +_FLAG_AUTO_DETECT = 0x08 + def _broadcast() -> None: """Tell running apps (browsers, etc.) that proxy settings changed.""" @@ -22,7 +47,136 @@ def _broadcast() -> None: pass -def set_system_proxy(host: str, port: int, bypass: str = "localhost;127.*;10.*;192.168.*;") -> None: +def _bump_counter(prev: bytes | None) -> int: + """Each blob write must bump byte offset 4 for WinINet to re-read it.""" + if not prev or len(prev) < 8: + return 1 + try: + return int.from_bytes(prev[4:8], "little") + 1 + except Exception: + return 1 + + +def _build_blob( + proxy: str, + bypass: str, + auto_config_url: str = "", + counter: int = 1, +) -> bytes: + """Build the ``DefaultConnectionSettings`` REG_BINARY value. + + Layout (little-endian): + u32 version (0x46) + u32 counter + u32 flags + u32 proxy_len; bytes proxy + u32 bypass_len; bytes bypass + u32 auto_config_len; bytes auto_config + bytes[32] padding (zeros) — WinINet expects trailing slack + """ + flags = _FLAG_DIRECT + if proxy: + flags |= _FLAG_MANUAL_PROXY + if auto_config_url: + flags |= _FLAG_AUTO_CONFIG + + p = proxy.encode("utf-8") + b = bypass.encode("utf-8") + a = auto_config_url.encode("utf-8") + + return ( + struct.pack(" int: + try: + with winreg.OpenKey( + winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH, 0, winreg.KEY_QUERY_VALUE + ) as key: + val, _ = winreg.QueryValueEx(key, "DefaultConnectionSettings") + return _bump_counter(val) + except OSError: + return 1 + + +def _write_conn_blob(proxy: str, bypass: str) -> None: + """Write per-connection binary so apps using the Connections key respect us.""" + counter = _read_blob_counter() + blob = _build_blob(proxy, bypass, "", counter) + try: + # The "Connections" key may not exist on freshly imaged Server installs. + key = winreg.CreateKeyEx( + winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH, 0, winreg.KEY_SET_VALUE + ) + with key: + winreg.SetValueEx(key, "DefaultConnectionSettings", 0, winreg.REG_BINARY, blob) + # SavedLegacySettings mirrors the same blob on some builds; keep them in sync. + winreg.SetValueEx(key, "SavedLegacySettings", 0, winreg.REG_BINARY, blob) + except OSError as e: + log.debug("Connections blob write failed: %s", e) + + +def _clear_conn_blob() -> None: + counter = _read_blob_counter() + blob = _build_blob("", "", "", counter) + try: + with winreg.OpenKey( + winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH, 0, winreg.KEY_SET_VALUE + ) as key: + winreg.SetValueEx(key, "DefaultConnectionSettings", 0, winreg.REG_BINARY, blob) + winreg.SetValueEx(key, "SavedLegacySettings", 0, winreg.REG_BINARY, blob) + except OSError: + pass + + +def _wipe_autoconfig(key: winreg.HKEYType) -> None: + """Remove WPAD / PAC overrides that would silently bypass our manual proxy.""" + for name in ("AutoConfigURL", "AutoDetect"): + try: + winreg.DeleteValue(key, name) + except OSError: + pass + + +def _set_winhttp_proxy(host: str, port: int, bypass: str) -> None: + """netsh winhttp covers .NET / services / Windows Update / some browsers.""" + try: + bypass_list = ( + bypass.replace(";", " ") + .replace("", "") + .strip() + ) or "" + subprocess.run( + ["netsh", "winhttp", "set", "proxy", + f"{host}:{port}", f"bypass-list={bypass_list}"], + capture_output=True, text=True, timeout=15, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except Exception as e: + log.debug("netsh winhttp set proxy failed: %s", e) + + +def _reset_winhttp_proxy() -> None: + try: + subprocess.run( + ["netsh", "winhttp", "reset", "proxy"], + capture_output=True, text=True, timeout=15, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except Exception: + pass + + +def set_system_proxy( + host: str, + port: int, + bypass: str = "localhost;127.*;10.*;192.168.*;", +) -> None: proxy = f"{host}:{port}" try: with winreg.OpenKey( @@ -31,10 +185,15 @@ def set_system_proxy(host: str, port: int, bypass: str = "localhost;127.*;10.*;1 winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1) winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy) winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass) - _broadcast() - log.info("System proxy set to %s (bypass: %s)", proxy, bypass) + _wipe_autoconfig(key) except OSError: - log.exception("Failed to set system proxy") + log.exception("Failed to set system proxy (HKCU root)") + return + + _write_conn_blob(proxy, bypass) + _set_winhttp_proxy(host, port, bypass) + _broadcast() + log.info("System proxy set %s (bypass=%s) — root + Connections + WinHTTP", proxy, bypass) def clear_system_proxy() -> None: @@ -43,11 +202,15 @@ def clear_system_proxy() -> None: winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE ) as key: winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0) - _broadcast() - log.info("System proxy cleared") + _wipe_autoconfig(key) except OSError: log.exception("Failed to clear system proxy") + _clear_conn_blob() + _reset_winhttp_proxy() + _broadcast() + log.info("System proxy cleared — root + Connections + WinHTTP") + def is_system_proxy_set() -> bool: try: diff --git a/proxy_chain_manager/validator.py b/proxy_chain_manager/validator.py index 8be332e..34b0223 100644 --- a/proxy_chain_manager/validator.py +++ b/proxy_chain_manager/validator.py @@ -10,8 +10,16 @@ import httpx log = logging.getLogger(__name__) -# Secondary fallback endpoints for IP resolution +# Secondary fallback endpoints for IP resolution. +# HTTP endpoints come first: many free proxies and older Windows Server +# environments fail HTTPS CONNECT through chained hops. Plain HTTP IP-check +# succeeds even when only forward-proxying (no CONNECT) works, so the chain's +# "last leg" exit-IP verification doesn't fail just because TLS can't tunnel. _IP_FALLBACKS = [ + "http://api.ipify.org/?format=json", + "http://checkip.amazonaws.com/", + "http://ip-api.com/json/", + "http://ifconfig.me/ip", "https://api.ipify.org?format=json", "https://httpbin.org/ip", "https://ifconfig.me/ip", @@ -144,14 +152,25 @@ async def check_chain_exit_ip( listen_proxy: str, check_url: str, timeout_seconds: float, + chain_hops: int = 3, ) -> str | None: """Query the IP-check URL through the local chain proxy. - Bounded total time — never stacks one slow request per fallback URL forever.""" - per = max(5.0, min(20.0, float(timeout_seconds))) - budget = max(15.0, min(60.0, float(timeout_seconds) * 2 + 5.0)) - urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url][:2] + + Bounded total time — never stacks one slow request per fallback URL forever. + Per-request timeout scales mildly with chain length so 5+ hop chains on + slower hardware (older Windows Server, low-spec VPS) don't time out on + cumulative TLS/CONNECT handshakes. + """ + hops = max(1, int(chain_hops)) + scale = 1.0 + max(0, hops - 3) * 0.35 + per = max(5.0, min(30.0, float(timeout_seconds) * scale)) + budget = max(20.0, min(90.0, float(timeout_seconds) * 2 * scale + 5.0)) + # Try more endpoints with HTTP first; if any hop blocks CONNECT, HTTPS + # IP-check would fail and the whole chain looks "dead" at the last step. + urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url][:4] log.debug( - "check_chain_exit_ip: budget=%.1fs per_req=%.1fs fallback_count=%d", + "check_chain_exit_ip: hops=%d budget=%.1fs per_req=%.1fs fallback_count=%d", + hops, budget, per, len(urls), diff --git a/proxy_chain_manager/vpn_detect.py b/proxy_chain_manager/vpn_detect.py index 848fb5f..02d1226 100644 --- a/proxy_chain_manager/vpn_detect.py +++ b/proxy_chain_manager/vpn_detect.py @@ -16,7 +16,15 @@ _VPN_ADAPTER_HINTS = ( "tailscale", "zerotier", "cisco anyconnect", "fortinet", "pulse secure", "globalprotect", "softether", "proton", "mullvad", "expressvpn", "surfshark", "private internet", "pia ", "windscribe", "hotspot shield", - "tunnel", "vpn", +) + +_NON_VPN_HINTS = ( + "wan miniport", + "microsoft kernel debug", + "isatap", + "teredo", + "loopback", + "pseudo-interface", ) # Executables to whitelist in kill-switch when present @@ -97,10 +105,45 @@ def detect_vpn() -> VpnStatus: """Inspect up network adapters for VPN/tunnel interfaces.""" out = _run_ps( "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " - "Select-Object -ExpandProperty Name" + "ForEach-Object { \"$($_.Name)|$($_.InterfaceDescription)\" }" ) if not out: - # Fallback: netsh + # Fallback for PowerShell 2.0 / older Windows Server: parse "wmic nic" + # then "netsh interface show interface". Both are locale-tolerant. + try: + r = subprocess.run( + ["wmic", "nic", "where", "NetEnabled=true", + "get", "NetConnectionID,Name", "/format:list"], + capture_output=True, + text=True, + timeout=12, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + blob = (r.stdout or "") + names: list[str] = [] + cur_id = "" + cur_name = "" + for ln in blob.splitlines(): + ln = ln.strip() + if not ln: + if cur_id: + names.append(f"{cur_id}|{cur_name}") + cur_id, cur_name = "", "" + continue + if ln.startswith("NetConnectionID="): + cur_id = ln.split("=", 1)[1].strip() + elif ln.startswith("Name="): + cur_name = ln.split("=", 1)[1].strip() + if cur_id: + names.append(f"{cur_id}|{cur_name}") + out = "\n".join(names) + except Exception: + out = "" + + if not out: + # Last-ditch: netsh. Locale-tolerant — match the connected/enabled state + # by extracting the interface name from the rightmost column. Older + # localized Server SKUs (es, de, fr, etc.) don't print literal "Enabled". try: r = subprocess.run( ["netsh", "interface", "show", "interface"], @@ -109,25 +152,42 @@ def detect_vpn() -> VpnStatus: timeout=10, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) - lines = (r.stdout or "").splitlines() + raw = (r.stdout or "") names = [] - for ln in lines[3:]: - parts = ln.split() - if len(parts) >= 4 and parts[0] == "Enabled": - names.append(" ".join(parts[3:])) + for ln in raw.splitlines(): + s = ln.rstrip() + if not s or s.startswith("-") or ":" in s.split(" ")[0]: + continue + parts = re.split(r"\s{2,}", s.strip()) + if len(parts) >= 4: + nm = parts[-1].strip() + if nm and nm.lower() not in ("interface name", "nombre de interfaz"): + names.append(f"{nm}|") out = "\n".join(names) except Exception: return VpnStatus() adapters: list[str] = [] + adapter_blob: list[str] = [] for line in out.splitlines(): - name = line.strip() + raw = line.strip() + if not raw: + continue + if "|" in raw: + name, desc = raw.split("|", 1) + else: + name, desc = raw, "" + name = name.strip() + desc = desc.strip() if name: adapters.append(name) + adapter_blob.append((name + " " + desc).strip()) hits: list[str] = [] - for name in adapters: - low = name.lower() + for i, name in enumerate(adapters): + low = adapter_blob[i].lower() + if any(h in low for h in _NON_VPN_HINTS): + continue if any(h in low for h in _VPN_ADAPTER_HINTS): hits.append(name) diff --git a/proxy_chain_manager/win_compat.py b/proxy_chain_manager/win_compat.py new file mode 100644 index 0000000..90c2e96 --- /dev/null +++ b/proxy_chain_manager/win_compat.py @@ -0,0 +1,90 @@ +"""Windows version / PowerShell capability probe. + +Used by privacy features to log a single clear reason when a function silently +no-ops on older Windows Server hosts (Server 2008 R2 ships PowerShell 2.0 which +does not have ``Get-NetAdapter`` / ``Set-NetAdapter`` / ``Disable-NetAdapterBinding`` +/ ``Get-DnsClientServerAddress`` / ``Rename-Computer``). +""" +from __future__ import annotations + +import logging +import platform +import subprocess +from dataclasses import dataclass +from functools import lru_cache + +log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class WinCompat: + release: str # e.g. "10", "2012ServerR2" + build: int # major build number + powershell_major: int # 0 = PowerShell not detected + has_net_cmdlets: bool # Get-NetAdapter etc. (PS 3.0+ on Server 2012+) + has_defender: bool # Add-MpPreference cmdlet exists + + def summary(self) -> str: + return ( + f"Windows release={self.release} build={self.build} " + f"PowerShell={self.powershell_major}.x " + f"NetAdapter cmdlets={'yes' if self.has_net_cmdlets else 'no'} " + f"Defender={'yes' if self.has_defender else 'no'}" + ) + + +def _powershell_major() -> int: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", + "$PSVersionTable.PSVersion.Major"], + capture_output=True, + text=True, + timeout=12, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + out = (r.stdout or "").strip().splitlines()[-1] if r.stdout else "" + return int(out) if out.isdigit() else 0 + except Exception: + return 0 + + +def _has_cmdlet(name: str) -> bool: + try: + r = subprocess.run( + ["powershell", "-NoProfile", "-NonInteractive", "-Command", + f"if (Get-Command {name} -ErrorAction SilentlyContinue) " + f"{{ 'yes' }} else {{ 'no' }}"], + capture_output=True, + text=True, + timeout=10, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return "yes" in (r.stdout or "").lower() + except Exception: + return False + + +@lru_cache(maxsize=1) +def probe() -> WinCompat: + """Cached host probe. Cheap on subsequent calls.""" + release = platform.release() + try: + build = int(platform.version().split(".")[-1]) + except Exception: + build = 0 + ps = _powershell_major() + # PowerShell >= 3.0 is the gate for the modern Net* cmdlets that all the + # privacy features rely on. Anything older (Server 2008 R2 RTM) only has + # PS 2.0 and needs WMI / netsh / ipconfig fallbacks. + has_net = ps >= 3 and _has_cmdlet("Get-NetAdapter") + has_def = _has_cmdlet("Add-MpPreference") + info = WinCompat( + release=release, + build=build, + powershell_major=ps, + has_net_cmdlets=has_net, + has_defender=has_def, + ) + log.info("WinCompat probe: %s", info.summary()) + return info diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py index 2e9ef3b..e915e3b 100644 --- a/tests/test_proxy_god.py +++ b/tests/test_proxy_god.py @@ -10,6 +10,7 @@ from proxy_chain_manager.config import ( Settings, merge_proxy_credentials, normalize_proxy_url, + parse_exit_proxy_input, redact_proxy_url, sanitize_settings, split_proxy_for_edit, @@ -87,6 +88,45 @@ class TestConfig(unittest.TestCase): self.assertTrue(changed) self.assertGreater(len(s2.sources), 0) + def test_parse_exit_host_port_only_defaults_to_socks5(self) -> None: + base, u, pw = parse_exit_proxy_input("1.2.3.4:1080") + self.assertEqual(base, "socks5://1.2.3.4:1080") + self.assertEqual(u, "") + self.assertEqual(pw, "") + + def test_parse_exit_colon_quadruple(self) -> None: + base, u, pw = parse_exit_proxy_input("1.2.3.4:1080:alice:s3cret") + self.assertEqual(base, "socks5://1.2.3.4:1080") + self.assertEqual(u, "alice") + self.assertEqual(pw, "s3cret") + + def test_parse_exit_password_with_colon_kept_intact(self) -> None: + base, u, pw = parse_exit_proxy_input("h:9:u:a:b:c") + self.assertEqual(base, "socks5://h:9") + self.assertEqual(u, "u") + self.assertEqual(pw, "a:b:c") + + def test_parse_exit_at_form_no_scheme(self) -> None: + base, u, pw = parse_exit_proxy_input("alice:secret@1.2.3.4:1080") + self.assertEqual(base, "socks5://1.2.3.4:1080") + self.assertEqual(u, "alice") + self.assertEqual(pw, "secret") + + def test_parse_exit_respects_explicit_scheme(self) -> None: + base, u, pw = parse_exit_proxy_input("http://x:y@9.9.9.9:8080") + self.assertEqual(base, "http://9.9.9.9:8080") + self.assertEqual(u, "x") + self.assertEqual(pw, "y") + + def test_parse_exit_whitespace_form(self) -> None: + base, u, pw = parse_exit_proxy_input("1.2.3.4:1080 alice s3cret") + self.assertEqual(base, "socks5://1.2.3.4:1080") + self.assertEqual(u, "alice") + self.assertEqual(pw, "s3cret") + + def test_parse_exit_empty(self) -> None: + self.assertEqual(parse_exit_proxy_input(" "), ("", "", "")) + def test_sanitize_clamps_extremes(self) -> None: s = Settings( chain_length=0,