138 lines
5.3 KiB
Python
138 lines
5.3 KiB
Python
"""Windows 系统代理模式：注册表切换、原状态保存与异常退出恢复。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import re
|
|
from typing import Callable
|
|
|
|
|
|
PROXY_REG = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Internet Settings"
|
|
BACKUP_KEY = "prev_proxy_state"
|
|
_VALUE_NAMES = ("ProxyEnable", "ProxyServer", "ProxyOverride", "AutoConfigURL")
|
|
|
|
|
|
def is_our_pac_url(value: str) -> bool:
|
|
return bool(re.fullmatch(r"http://127\.0\.0\.1:\d+/proxy\.pac", str(value).strip()))
|
|
|
|
|
|
def notify_proxy_change() -> None:
|
|
"""通知 WinINET 和顶层窗口重新读取系统代理。"""
|
|
try:
|
|
ctypes.windll.Wininet.InternetSetOptionW(None, 39, None, 0)
|
|
ctypes.windll.Wininet.InternetSetOptionW(None, 37, None, 0)
|
|
result = ctypes.c_ulong()
|
|
ctypes.windll.user32.SendMessageTimeoutW(
|
|
0xFFFF, 0x001A, 0, ctypes.c_wchar_p("InternetSettings"),
|
|
0x0002, 5000, ctypes.byref(result))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class SystemProxyMode:
|
|
def __init__(self, *, run: Callable, settings_load: Callable,
|
|
settings_save: Callable, log: Callable,
|
|
endpoint: Callable[[], tuple[str, int]],
|
|
proxy_override: Callable[[bool], str]):
|
|
self.run = run
|
|
self.settings_load = settings_load
|
|
self.settings_save = settings_save
|
|
self.log = log
|
|
self.endpoint = endpoint
|
|
self.proxy_override = proxy_override
|
|
self._previous = None
|
|
|
|
def reg_query(self, name: str):
|
|
result = self.run(["reg", "query", PROXY_REG, "/v", name], warn=False)
|
|
if result.returncode:
|
|
return None
|
|
for line in (result.stdout or "").splitlines():
|
|
parts = line.strip().split(None, 2)
|
|
if len(parts) >= 3 and parts[0].lower() == name.lower():
|
|
return parts[1], parts[2]
|
|
return None
|
|
|
|
def _backup(self):
|
|
if self._previous is not None:
|
|
return
|
|
state = {name: list(value) if (value := self.reg_query(name)) else None
|
|
for name in _VALUE_NAMES}
|
|
self._previous = state
|
|
try:
|
|
self.settings_save({**self.settings_load(), BACKUP_KEY: state})
|
|
except Exception:
|
|
pass
|
|
|
|
def _restore(self):
|
|
state = self._previous
|
|
if state is None:
|
|
state = self.settings_load().get(BACKUP_KEY)
|
|
if isinstance(state, dict):
|
|
for name in _VALUE_NAMES:
|
|
value = state.get(name)
|
|
if value:
|
|
self.run(["reg", "add", PROXY_REG, "/v", name, "/t", value[0],
|
|
"/d", value[1], "/f"], warn=False)
|
|
else:
|
|
self.run(["reg", "delete", PROXY_REG, "/v", name, "/f"], warn=False)
|
|
self.log("已恢复开启代理前的系统代理设置")
|
|
else:
|
|
for name in ("ProxyServer", "ProxyOverride", "AutoConfigURL"):
|
|
self.run(["reg", "delete", PROXY_REG, "/v", name, "/f"], warn=False)
|
|
self.run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t",
|
|
"REG_DWORD", "/d", "0", "/f"])
|
|
self._previous = None
|
|
try:
|
|
data = self.settings_load()
|
|
if BACKUP_KEY in data:
|
|
del data[BACKUP_KEY]
|
|
self.settings_save(data)
|
|
except Exception:
|
|
pass
|
|
|
|
def enable(self, pac_url=None, bypass_lan=False):
|
|
self._backup()
|
|
if pac_url:
|
|
self.run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t",
|
|
"REG_DWORD", "/d", "0", "/f"])
|
|
for name in ("ProxyServer", "ProxyOverride"):
|
|
self.run(["reg", "delete", PROXY_REG, "/v", name, "/f"], warn=False)
|
|
self.run(["reg", "add", PROXY_REG, "/v", "AutoConfigURL", "/t",
|
|
"REG_SZ", "/d", pac_url, "/f"])
|
|
self.log("系统代理已设置 (PAC)")
|
|
else:
|
|
host, port = self.endpoint()
|
|
address = f"{host}:{port}"
|
|
self.run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t",
|
|
"REG_DWORD", "/d", "1", "/f"])
|
|
self.run(["reg", "add", PROXY_REG, "/v", "ProxyServer", "/t",
|
|
"REG_SZ", "/d", address, "/f"])
|
|
self.run(["reg", "add", PROXY_REG, "/v", "ProxyOverride", "/t",
|
|
"REG_SZ", "/d", self.proxy_override(bypass_lan), "/f"])
|
|
self.log(f"系统代理已设置: {address}")
|
|
notify_proxy_change()
|
|
|
|
def disable(self):
|
|
self._restore()
|
|
notify_proxy_change()
|
|
|
|
def is_on(self):
|
|
host, port = self.endpoint()
|
|
enabled = self.reg_query("ProxyEnable")
|
|
try:
|
|
fixed_enabled = bool(enabled and int(str(enabled[1]).strip(), 0) == 1)
|
|
except (TypeError, ValueError):
|
|
fixed_enabled = False
|
|
if fixed_enabled:
|
|
server = self.reg_query("ProxyServer")
|
|
return bool(server and (server[1] or "").strip() == f"{host}:{port}")
|
|
pac = self.reg_query("AutoConfigURL")
|
|
return bool(pac) and is_our_pac_url(pac[1] or "")
|
|
|
|
def is_stale(self):
|
|
host, port = self.endpoint()
|
|
server = self.reg_query("ProxyServer")
|
|
pac = self.reg_query("AutoConfigURL")
|
|
return (bool(server) and (server[1] or "").strip() == f"{host}:{port}") or \
|
|
(bool(pac) and is_our_pac_url(pac[1] or ""))
|