Compare commits

..

2 Commits

Author SHA1 Message Date
Indiana Holmes
7657a37855 Fix Windows Server proxy: HKLM mirror, ProxySettingsPerUser fix, policy detect
On Server / GPO baselines, ProxySettingsPerUser=0 makes WinINet ignore HKCU entirely. When elevated, force the flag to 1 and mirror proxy values to HKLM root + Connections blob. Detect Group Policy proxy locks and surface them so the user knows browsers will keep the policy value. Add diagnose_system_proxy() logged on every chain engage.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-16 18:08:32 -07:00
Indiana Holmes
8f012402a6 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>
2026-05-16 18:03:30 -07:00
10 changed files with 792 additions and 43 deletions

View File

@@ -24,6 +24,7 @@ from .config import (
load_settings, load_settings,
merge_proxy_credentials, merge_proxy_credentials,
normalize_proxy_url, normalize_proxy_url,
parse_exit_proxy_input,
redact_proxy_url, redact_proxy_url,
save_settings, save_settings,
split_proxy_for_edit, split_proxy_for_edit,
@@ -490,7 +491,7 @@ def main() -> None:
exit_test_dot.pack(side="left", padx=(8, 4)) exit_test_dot.pack(side="left", padx=(8, 4))
exit_proxy_entry = ctk.CTkEntry( exit_proxy_entry = ctk.CTkEntry(
exit_fix_frame, 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), font=("Consolas", 10),
fg_color=BG, fg_color=BG,
border_color=ACCENT, border_color=ACCENT,
@@ -527,13 +528,49 @@ def main() -> None:
exit_btn_row = ctk.CTkFrame(exit_fix_frame, fg_color="transparent") exit_btn_row = ctk.CTkFrame(exit_fix_frame, fg_color="transparent")
exit_btn_row.pack(fill="x", padx=10, pady=(0, 4)) 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: def _exit_url() -> str:
raw = exit_proxy_entry.get().strip() """Build the final exit-proxy URL.
if not raw:
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 "" return ""
if "://" not in raw: user = (exit_user_entry.get() or "").strip() or parsed_user
raw = "http://" + raw pw = (exit_pass_entry.get() or "").strip() or parsed_pw
return merge_proxy_credentials(raw, exit_user_entry.get(), exit_pass_entry.get()) return merge_proxy_credentials(base, user, pw)
def _test_exit() -> None: def _test_exit() -> None:
url = _exit_url() url = _exit_url()
@@ -567,8 +604,11 @@ def main() -> None:
ctk.CTkLabel( ctk.CTkLabel(
exit_fix_frame, exit_fix_frame,
text="Appended after your chain hops when running. Leave empty to use last chain hop as exit.\n" text=(
"Auth: User/Pass fields or user:pass@host in the URL.", "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), font=(FONT, 9),
text_color=TEXT2, text_color=TEXT2,
justify="left", justify="left",
@@ -1247,11 +1287,7 @@ def main() -> None:
obfuscation_mode=mode_var.get(), obfuscation_mode=mode_var.get(),
use_pinned_chain=bool(use_manual_var.get()), use_pinned_chain=bool(use_manual_var.get()),
pinned_chain=list(manual_chain), pinned_chain=list(manual_chain),
manual_exit_proxy=merge_proxy_credentials( manual_exit_proxy=_exit_url(),
exit_proxy_entry.get(),
exit_user_entry.get(),
exit_pass_entry.get(),
),
health_check_seconds=min(3600, max(10, int(entries["health"].get().strip()))), health_check_seconds=min(3600, max(10, int(entries["health"].get().strip()))),
full_refresh_seconds=min(86400, max(60, int(entries["refresh"].get().strip()))), full_refresh_seconds=min(86400, max(60, int(entries["refresh"].get().strip()))),
validation_concurrency=max(1, int(entries["conc"].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("/") 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: def redact_proxy_url(url: str) -> str:
"""Same URL shape for logs/UI, with credentials replaced by ``***``.""" """Same URL shape for logs/UI, with credentials replaced by ``***``."""
t = (url or "").strip() t = (url or "").strip()

View File

@@ -11,6 +11,7 @@ from dataclasses import dataclass, field
from .firewall import is_admin from .firewall import is_admin
from .mac_spoof import list_nics from .mac_spoof import list_nics
from .win_compat import probe as _win_probe
log = logging.getLogger(__name__) 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).""" """Disable IPv6 binding on up physical adapters. Returns (adapter names, log lines)."""
if not is_admin(): if not is_admin():
return [], ["IPv6 disable skipped (not 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 = ( script = (
"Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | "
"ForEach-Object { $_.Name }" "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]: def enable_ipv6_on_adapters(adapters: list[str]) -> list[str]:
if not is_admin() or not adapters: if not is_admin() or not adapters:
return [] return []
if not _win_probe().has_net_cmdlets:
return []
logs: list[str] = [] logs: list[str] = []
for name in adapters: for name in adapters:
cmd = ( cmd = (

View File

@@ -8,6 +8,7 @@ import subprocess
from dataclasses import dataclass from dataclasses import dataclass
from .firewall import is_admin from .firewall import is_admin
from .win_compat import probe as _win_probe
log = logging.getLogger(__name__) 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]: 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 = ( script = (
"Get-NetAdapter | Where-Object { $_.Status -ne 'Disabled' } | " "Get-NetAdapter | Where-Object { $_.Status -ne 'Disabled' } | "
"Select-Object Name, MacAddress, InterfaceDescription | " "Select-Object Name, MacAddress, InterfaceDescription | "
@@ -44,7 +52,7 @@ def list_nics() -> list[NicMac]:
) )
code, out, _ = _run(["powershell", "-NoProfile", "-Command", script]) code, out, _ = _run(["powershell", "-NoProfile", "-Command", script])
if code != 0 or not out.strip(): if code != 0 or not out.strip():
return [] return _list_nics_wmic()
import json import json
try: try:
@@ -71,9 +79,46 @@ def random_mac() -> str:
return ":".join(f"{x:02X}" for x in b) 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]: def set_mac(adapter: str, mac: str) -> tuple[bool, str]:
if not is_admin(): if not is_admin():
return False, "Administrator required to change MAC." 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() mac = mac.replace("-", ":").upper()
if not _MAC_RE.match(mac.replace(":", "-")): if not _MAC_RE.match(mac.replace(":", "-")):
return False, f"Invalid MAC: {mac}" return False, f"Invalid MAC: {mac}"

View File

@@ -30,9 +30,16 @@ 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 .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 .leak_detect import is_chain_leak, leak_reason
from .mac_spoof import restore_macs, spoof_all_physical from .mac_spoof import restore_macs, spoof_all_physical
from .sysproxy import clear_system_proxy, set_system_proxy from .sysproxy import (
clear_system_proxy,
detect_policy_overrides,
diagnose_system_proxy,
is_system_proxy_set,
set_system_proxy,
)
from .validator import check_chain_exit_ip, get_direct_ip, validate_proxies from .validator import check_chain_exit_ip, get_direct_ip, validate_proxies
from .vpn_detect import VpnStatus, detect_vpn from .vpn_detect import VpnStatus, detect_vpn
from .win_compat import probe as _win_probe
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -89,6 +96,17 @@ class ChainService:
def start(self) -> None: def start(self) -> None:
if self._thread and self._thread.is_alive(): if self._thread and self._thread.is_alive():
return 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._stop.clear()
self._thread = threading.Thread( self._thread = threading.Thread(
target=self._run_thread, name="ChainService", daemon=True target=self._run_thread, name="ChainService", daemon=True
@@ -419,7 +437,9 @@ class ChainService:
"text": f"Exit IP check through local proxy (≤{int(timeout * 2 + 5)}s)…", "text": f"Exit IP check through local proxy (≤{int(timeout * 2 + 5)}s)…",
}) })
t0 = time.monotonic() 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") log.debug("Initial exit IP check took %.2fs → %s", time.monotonic() - t0, exit_ip or "none")
if not exit_ip: if not exit_ip:
@@ -448,12 +468,34 @@ class ChainService:
self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip}) self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip})
self._notify({"type": "log", "text": f"✓ Chain healthy — Exit IP: {exit_ip}"}) self._notify({"type": "log", "text": f"✓ Chain healthy — Exit IP: {exit_ip}"})
self._notify({"type": "phase", "phase": "running"}) self._notify({"type": "phase", "phase": "running"})
# Pre-flight: surface Group Policy locks (they will override us).
pol_before = detect_policy_overrides()
if pol_before:
self._notify({"type": "log", "text": (
f"Group Policy proxy lock detected ({len(pol_before)} entries) — "
"these BEAT our settings. Browsers will keep the policy proxy "
"(or DIRECT) until those keys are removed."
)})
for p in pol_before[:3]:
self._notify({"type": "log", "text": f" ! {p}"})
set_system_proxy( set_system_proxy(
self._settings.local_host, self._settings.local_host,
self._settings.local_port, self._settings.local_port,
self._settings.proxy_bypass, 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"(HKCU + HKLM(adm) + Connections + WinHTTP) "
f"{'OK' if applied else 'FAILED — registry write rejected'}"
),
})
# Post-flight diagnostic — every layer's actual state.
for ln in diagnose_system_proxy():
self._notify({"type": "log", "text": f" proxy: {ln}"})
# ── Health monitor loop ─────────────────────────────────────────────── # ── Health monitor loop ───────────────────────────────────────────────
hc = int(self._settings.health_check_seconds) hc = int(self._settings.health_check_seconds)
@@ -470,7 +512,9 @@ class ChainService:
self._notify({"type": "log", "text": "Health check..."}) self._notify({"type": "log", "text": "Health check..."})
t1 = time.monotonic() 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") 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): if is_chain_leak(exit_ip, real_ip, self._vpn.active):

View File

@@ -1,16 +1,62 @@
"""Set / clear the Windows system-wide HTTP proxy via the registry + WinINet broadcast.""" """Set / clear the Windows system-wide HTTP proxy.
Layers 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.
3. ``HKLM\\\\Internet Settings`` mirror (Admin only) — required on Server when
``ProxySettingsPerUser=0`` is set by Group Policy / baseline image.
4. ``netsh winhttp set proxy`` — .NET, Windows Update, services and
headless browser components that bypass WinINet.
We also actively detect and warn about:
- ``ProxySettingsPerUser=0`` → HKCU is ignored; we fix it when Admin.
- Group Policy proxy keys → those override us; we surface the path so the
user can verify or remove them.
- ``AutoConfigURL`` / ``AutoDetect`` (WPAD) → wiped on engage.
"""
from __future__ import annotations from __future__ import annotations
import ctypes import ctypes
import logging import logging
import struct
import subprocess
import winreg import winreg
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
_KEY_PATH = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings" _KEY_PATH = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
_CONN_KEY_PATH = (
r"Software\Microsoft\Windows\CurrentVersion\Internet Settings\Connections"
)
_HKLM_INET_PATH = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Internet Settings"
_HKLM_CONN_PATH = (
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Internet Settings\Connections"
)
_POLICY_PATHS = (
(winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Policies\Microsoft\Windows\CurrentVersion\Internet Settings"),
(winreg.HKEY_CURRENT_USER,
r"SOFTWARE\Policies\Microsoft\Windows\CurrentVersion\Internet Settings"),
(winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Policies\Microsoft\Internet Explorer\Control Panel"),
)
_SETTINGS_CHANGED = 39 _SETTINGS_CHANGED = 39
_REFRESH = 37 _REFRESH = 37
_FLAG_DIRECT = 0x01
_FLAG_MANUAL_PROXY = 0x02
_FLAG_AUTO_CONFIG = 0x04
_FLAG_AUTO_DETECT = 0x08
def _is_admin() -> bool:
try:
return ctypes.windll.shell32.IsUserAnAdmin() != 0 # type: ignore[attr-defined]
except Exception:
return False
def _broadcast() -> None: def _broadcast() -> None:
"""Tell running apps (browsers, etc.) that proxy settings changed.""" """Tell running apps (browsers, etc.) that proxy settings changed."""
@@ -22,8 +68,281 @@ def _broadcast() -> None:
pass 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:
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:
"""REG_BINARY layout for DefaultConnectionSettings (little-endian)."""
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(hive: int, path: str) -> int:
try:
with winreg.OpenKey(hive, 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(hive: int, path: str, proxy: str, bypass: str) -> bool:
counter = _read_blob_counter(hive, path)
blob = _build_blob(proxy, bypass, "", counter)
try:
key = winreg.CreateKeyEx(hive, path, 0, winreg.KEY_SET_VALUE)
with key:
winreg.SetValueEx(key, "DefaultConnectionSettings", 0, winreg.REG_BINARY, blob)
winreg.SetValueEx(key, "SavedLegacySettings", 0, winreg.REG_BINARY, blob)
return True
except OSError as e:
log.debug("Connections blob write failed (%s): %s", path, e)
return False
def _clear_conn_blob(hive: int, path: str) -> None:
counter = _read_blob_counter(hive, path)
blob = _build_blob("", "", "", counter)
try:
with winreg.OpenKey(hive, 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:
for name in ("AutoConfigURL", "AutoDetect"):
try:
winreg.DeleteValue(key, name)
except OSError:
pass
def _set_winhttp_proxy(host: str, port: int, bypass: str) -> None:
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 _read_proxy_settings_per_user() -> int | None:
"""``HKLM\\\\Internet Settings\\ProxySettingsPerUser``.
When 0, Windows ignores HKCU proxy values and reads only HKLM. Set on many
Server images and AD policy baselines — the #1 cause of "set the proxy but
the browser still goes direct" on Server.
"""
try:
with winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_QUERY_VALUE
) as key:
val, _ = winreg.QueryValueEx(key, "ProxySettingsPerUser")
return int(val)
except OSError:
return None
def _fix_proxy_settings_per_user() -> bool:
"""Force ``ProxySettingsPerUser=1`` (per-user proxy honored). Admin only."""
if not _is_admin():
return False
try:
with winreg.CreateKeyEx(
winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
) as key:
winreg.SetValueEx(key, "ProxySettingsPerUser", 0, winreg.REG_DWORD, 1)
log.info("HKLM\\\\ProxySettingsPerUser → 1 (was %s)", _read_proxy_settings_per_user())
return True
except OSError as e:
log.warning("Could not fix ProxySettingsPerUser: %s", e)
return False
def _set_hklm_root(proxy: str, bypass: str) -> bool:
"""Mirror the per-user keys at HKLM so machine-scope WinINet reads honor us."""
if not _is_admin():
return False
try:
with winreg.CreateKeyEx(
winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
) as key:
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)
_wipe_autoconfig(key)
return True
except OSError as e:
log.warning("HKLM root proxy write failed: %s", e)
return False
def _clear_hklm_root() -> None:
if not _is_admin():
return
try:
with winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, 0, winreg.KEY_SET_VALUE
) as key:
winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)
_wipe_autoconfig(key)
except OSError:
pass
def detect_policy_overrides() -> list[str]:
"""Return human-readable paths of Group Policy proxy locks that beat us.
Anything we find here will silently override our manual proxy. We do *not*
delete policy keys (admins set those deliberately) — we just surface them.
"""
hits: list[str] = []
proxy_value_names = ("ProxyServer", "ProxyEnable", "AutoConfigURL", "ProxyOverride")
for hive, path in _POLICY_PATHS:
try:
with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
for n in proxy_value_names:
try:
winreg.QueryValueEx(key, n)
hive_name = "HKLM" if hive == winreg.HKEY_LOCAL_MACHINE else "HKCU"
hits.append(f"{hive_name}\\{path}\\{n}")
except OSError:
continue
except OSError:
continue
return hits
def _read_str(hive: int, path: str, name: str) -> str:
try:
with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
val, _ = winreg.QueryValueEx(key, name)
return str(val)
except OSError:
return ""
def _read_dword(hive: int, path: str, name: str) -> int | None:
try:
with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
val, _ = winreg.QueryValueEx(key, name)
return int(val)
except OSError:
return None
def _read_winhttp_proxy() -> str:
try:
r = subprocess.run(
["netsh", "winhttp", "show", "proxy"],
capture_output=True, text=True, timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
for ln in (r.stdout or "").splitlines():
if "proxy" in ln.lower() and ":" in ln:
return ln.strip()
return (r.stdout or "").strip().splitlines()[-1] if r.stdout else ""
except Exception:
return ""
def diagnose_system_proxy() -> list[str]:
"""Return a list of human-readable lines covering every place proxy can
be configured. Logged on each engage so "chain green but browser leaks"
is debuggable without RegEdit gymnastics.
"""
lines: list[str] = []
per_user = _read_proxy_settings_per_user()
per_user_msg = (
"1 (HKCU honored)" if per_user == 1
else "0 (HKCU IGNORED — Server lock!)" if per_user == 0
else "(unset → defaults to HKCU)"
)
lines.append(f"ProxySettingsPerUser = {per_user_msg}")
en_u = _read_dword(winreg.HKEY_CURRENT_USER, _KEY_PATH, "ProxyEnable")
ps_u = _read_str(winreg.HKEY_CURRENT_USER, _KEY_PATH, "ProxyServer")
lines.append(f"HKCU Enable={en_u} Server='{ps_u}'")
en_m = _read_dword(winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, "ProxyEnable")
ps_m = _read_str(winreg.HKEY_LOCAL_MACHINE, _HKLM_INET_PATH, "ProxyServer")
lines.append(f"HKLM Enable={en_m} Server='{ps_m}'")
ac_u = _read_str(winreg.HKEY_CURRENT_USER, _KEY_PATH, "AutoConfigURL")
ad_u = _read_dword(winreg.HKEY_CURRENT_USER, _KEY_PATH, "AutoDetect")
if ac_u or ad_u:
lines.append(f"WPAD/PAC: AutoConfigURL='{ac_u}' AutoDetect={ad_u}")
winhttp = _read_winhttp_proxy()
if winhttp:
lines.append(f"WinHTTP: {winhttp}")
pol = detect_policy_overrides()
if pol:
lines.append(f"Policy overrides (these BEAT us): {len(pol)} entries")
for p in pol[:5]:
lines.append(f" ! {p}")
if len(pol) > 5:
lines.append(f" ! …and {len(pol) - 5} more")
else:
lines.append("Policy overrides: none")
return lines
def set_system_proxy(
host: str,
port: int,
bypass: str = "localhost;127.*;10.*;192.168.*;<local>",
) -> None:
proxy = f"{host}:{port}" proxy = f"{host}:{port}"
# HKCU root
try: try:
with winreg.OpenKey( with winreg.OpenKey(
winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE
@@ -31,10 +350,40 @@ 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, "ProxyEnable", 0, winreg.REG_DWORD, 1)
winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy) winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy)
winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass) winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass)
_broadcast() _wipe_autoconfig(key)
log.info("System proxy set to %s (bypass: %s)", proxy, bypass)
except OSError: except OSError:
log.exception("Failed to set system proxy") log.exception("Failed to set system proxy (HKCU root)")
return
_write_conn_blob(winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH, proxy, bypass)
# If ProxySettingsPerUser=0 (Server / GPO baseline), HKCU is ignored.
# When Admin, fix the flag AND mirror to HKLM so the proxy actually applies.
per_user = _read_proxy_settings_per_user()
if per_user == 0:
log.warning(
"ProxySettingsPerUser=0 detected — Windows is ignoring per-user "
"proxy. Attempting Admin fix (HKLM mirror)."
)
if _is_admin():
_fix_proxy_settings_per_user()
_set_hklm_root(proxy, bypass)
_write_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH, proxy, bypass)
else:
log.warning(
"Not running as Admin — cannot fix ProxySettingsPerUser. "
"Re-launch elevated for proxy to apply on this Server image."
)
# On Admin runs always mirror HKLM root too. Harmless when PerUser=1 and
# correct when PerUser=0 or absent.
if _is_admin():
_set_hklm_root(proxy, bypass)
_write_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH, proxy, bypass)
_set_winhttp_proxy(host, port, bypass)
_broadcast()
log.info("System proxy set %s (bypass=%s) — HKCU + HKLM(adm) + Connections + WinHTTP", proxy, bypass)
def clear_system_proxy() -> None: def clear_system_proxy() -> None:
@@ -43,11 +392,18 @@ def clear_system_proxy() -> None:
winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE winreg.HKEY_CURRENT_USER, _KEY_PATH, 0, winreg.KEY_SET_VALUE
) as key: ) as key:
winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0) winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)
_broadcast() _wipe_autoconfig(key)
log.info("System proxy cleared")
except OSError: except OSError:
log.exception("Failed to clear system proxy") log.exception("Failed to clear system proxy")
_clear_conn_blob(winreg.HKEY_CURRENT_USER, _CONN_KEY_PATH)
if _is_admin():
_clear_hklm_root()
_clear_conn_blob(winreg.HKEY_LOCAL_MACHINE, _HKLM_CONN_PATH)
_reset_winhttp_proxy()
_broadcast()
log.info("System proxy cleared — HKCU + HKLM(adm) + Connections + WinHTTP")
def is_system_proxy_set() -> bool: def is_system_proxy_set() -> bool:
try: try:

View File

@@ -10,8 +10,16 @@ import httpx
log = logging.getLogger(__name__) 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 = [ _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://api.ipify.org?format=json",
"https://httpbin.org/ip", "https://httpbin.org/ip",
"https://ifconfig.me/ip", "https://ifconfig.me/ip",
@@ -144,14 +152,25 @@ async def check_chain_exit_ip(
listen_proxy: str, listen_proxy: str,
check_url: str, check_url: str,
timeout_seconds: float, timeout_seconds: float,
chain_hops: int = 3,
) -> str | None: ) -> str | None:
"""Query the IP-check URL through the local chain proxy. """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))) Bounded total time — never stacks one slow request per fallback URL forever.
budget = max(15.0, min(60.0, float(timeout_seconds) * 2 + 5.0)) Per-request timeout scales mildly with chain length so 5+ hop chains on
urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url][:2] 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( 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, budget,
per, per,
len(urls), len(urls),

View File

@@ -16,7 +16,15 @@ _VPN_ADAPTER_HINTS = (
"tailscale", "zerotier", "cisco anyconnect", "fortinet", "pulse secure", "tailscale", "zerotier", "cisco anyconnect", "fortinet", "pulse secure",
"globalprotect", "softether", "proton", "mullvad", "expressvpn", "globalprotect", "softether", "proton", "mullvad", "expressvpn",
"surfshark", "private internet", "pia ", "windscribe", "hotspot shield", "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 # Executables to whitelist in kill-switch when present
@@ -97,10 +105,45 @@ def detect_vpn() -> VpnStatus:
"""Inspect up network adapters for VPN/tunnel interfaces.""" """Inspect up network adapters for VPN/tunnel interfaces."""
out = _run_ps( out = _run_ps(
"Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | " "Get-NetAdapter | Where-Object { $_.Status -eq 'Up' } | "
"Select-Object -ExpandProperty Name" "ForEach-Object { \"$($_.Name)|$($_.InterfaceDescription)\" }"
) )
if not out: 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: try:
r = subprocess.run( r = subprocess.run(
["netsh", "interface", "show", "interface"], ["netsh", "interface", "show", "interface"],
@@ -109,25 +152,42 @@ def detect_vpn() -> VpnStatus:
timeout=10, timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) )
lines = (r.stdout or "").splitlines() raw = (r.stdout or "")
names = [] names = []
for ln in lines[3:]: for ln in raw.splitlines():
parts = ln.split() s = ln.rstrip()
if len(parts) >= 4 and parts[0] == "Enabled": if not s or s.startswith("-") or ":" in s.split(" ")[0]:
names.append(" ".join(parts[3:])) 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) out = "\n".join(names)
except Exception: except Exception:
return VpnStatus() return VpnStatus()
adapters: list[str] = [] adapters: list[str] = []
adapter_blob: list[str] = []
for line in out.splitlines(): 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: if name:
adapters.append(name) adapters.append(name)
adapter_blob.append((name + " " + desc).strip())
hits: list[str] = [] hits: list[str] = []
for name in adapters: for i, name in enumerate(adapters):
low = name.lower() 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): if any(h in low for h in _VPN_ADAPTER_HINTS):
hits.append(name) 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, Settings,
merge_proxy_credentials, merge_proxy_credentials,
normalize_proxy_url, normalize_proxy_url,
parse_exit_proxy_input,
redact_proxy_url, redact_proxy_url,
sanitize_settings, sanitize_settings,
split_proxy_for_edit, split_proxy_for_edit,
@@ -87,6 +88,45 @@ class TestConfig(unittest.TestCase):
self.assertTrue(changed) self.assertTrue(changed)
self.assertGreater(len(s2.sources), 0) 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: def test_sanitize_clamps_extremes(self) -> None:
s = Settings( s = Settings(
chain_length=0, chain_length=0,