feat(lite): merge TUN mode with IPv6 routing

lite/rrpp_core/system_proxy_mode.py

@@ -0,0 +1,137 @@
+"""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 ""))