Compare commits
2 Commits
8bd8d4267f
...
7657a37855
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7657a37855 | ||
|
|
8f012402a6 |
@@ -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())),
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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 = (
|
||||
|
||||
@@ -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}"
|
||||
|
||||
@@ -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 .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,
|
||||
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 .vpn_detect import VpnStatus, detect_vpn
|
||||
from .win_compat import probe as _win_probe
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -89,6 +96,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 +437,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:
|
||||
@@ -448,12 +468,34 @@ class ChainService:
|
||||
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": "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(
|
||||
self._settings.local_host,
|
||||
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"(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 ───────────────────────────────────────────────
|
||||
hc = int(self._settings.health_check_seconds)
|
||||
@@ -470,7 +512,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):
|
||||
|
||||
@@ -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
|
||||
|
||||
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"
|
||||
)
|
||||
_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
|
||||
_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:
|
||||
"""Tell running apps (browsers, etc.) that proxy settings changed."""
|
||||
@@ -22,8 +68,281 @@ 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:
|
||||
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}"
|
||||
|
||||
# HKCU root
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
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, "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(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:
|
||||
@@ -43,11 +392,18 @@ 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(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:
|
||||
try:
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
90
proxy_chain_manager/win_compat.py
Normal file
90
proxy_chain_manager/win_compat.py
Normal 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
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user