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 <cursoragent@cursor.com>
This commit is contained in:
Indiana Holmes
2026-05-16 18:03:30 -07:00
parent 8bd8d4267f
commit 8f012402a6
10 changed files with 579 additions and 43 deletions

View File

@@ -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("<KeyRelease>", _auto_split_exit_paste)
exit_proxy_entry.bind("<<Paste>>", lambda e: exit_proxy_entry.after(1, _auto_split_exit_paste))
exit_proxy_entry.bind("<FocusOut>", _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())),

View File

@@ -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()

View File

@@ -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 = (

View File

@@ -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}"

View File

@@ -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):

View File

@@ -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.*;<local>") -> 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("<III", 0x46, counter, flags)
+ struct.pack("<I", len(p)) + p
+ struct.pack("<I", len(b)) + b
+ struct.pack("<I", len(a)) + a
+ b"\x00" * 32
)
def _read_blob_counter() -> 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("<local>", "<local>")
.strip()
) or "<local>"
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.*;<local>",
) -> 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:

View File

@@ -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),

View File

@@ -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)

View File

@@ -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

View File

@@ -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,