# -*- coding: utf-8 -*- """ VPN 分流助手（GUI 版） 功能：本地流量走本地默认网关；仅指定网段走指定 L2TP VPN（split tunneling）。 运行：python vpn_split_gui.py （非管理员会自动弹 UAC 提权重启） 测试：python vpn_split_gui.py --test （只跑核心检测，不开窗口） """ import ctypes import json import os import queue import re import subprocess import sys import threading import time import tkinter as tk from tkinter import ttk, messagebox, scrolledtext # ==================== 默认变量（GUI 内可改，保存到 config.json） ==================== DEFAULTS = { "vpn_name": "腾讯云", # 拨号列表里的 VPN 连接名 "target_cidr": "58.220.83.0/24", # 走 VPN 的目标网段 "route_metric": 1, # 路由优先级，越小越优先 "username": "", # 留空 = 用系统已保存凭据 "password": "", "auto_dial": True, # 一键设置时自动拨号 "keepalive": False, # 断线自动重拨并补路由 "disable_vpn_ipv6": False, # 关闭 VPN 适配器 IPv6，防泄漏 "is_all_user": True, # 全局连接（所有用户可见） } CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "vpn_split_config.json") KEEPALIVE_INTERVAL = 10 # 秒 # ============================================================================== # ------------------------------- 基础执行函数 ------------------------------- def run_cmd(args, timeout=90): """执行外部命令，返回 (stdout, returncode)。中文控制台输出按 ANSI/GBK 解码。""" try: p = subprocess.run(args, capture_output=True, timeout=timeout, creationflags=subprocess.CREATE_NO_WINDOW) out = p.stdout.decode("mbcs", "replace") + p.stderr.decode("mbcs", "replace") return out, p.returncode except Exception as e: return "执行异常: %s" % e, -1 def run_ps(script, timeout=90): """执行 PowerShell，统一 UTF-8 输出。返回 (stdout, returncode)。""" full = '[Console]::OutputEncoding=[System.Text.Encoding]::UTF8; ' + script try: p = subprocess.run( ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", full], capture_output=True, timeout=timeout, creationflags=subprocess.CREATE_NO_WINDOW) out = p.stdout.decode("utf-8", "replace") + p.stderr.decode("utf-8", "replace") return out, p.returncode except Exception as e: return "执行异常: %s" % e, -1 def is_admin(): try: return bool(ctypes.windll.shell32.IsUserAnAdmin()) except Exception: return False def elevate(): """以管理员身份重启自身。""" params = '"%s"' % os.path.abspath(sys.argv[0]) if "--test" in sys.argv: params += " --test" ret = ctypes.windll.shell32.ShellExecuteW( None, "runas", sys.executable, params, None, 1) if ret <= 32: # pythonw 无控制台时用 ShellExecute 直接启动可能失败，退回 python.exe ctypes.windll.shell32.ShellExecuteW( None, "runas", sys.executable.replace("pythonw.exe", "python.exe"), params, None, 1) sys.exit(0) # ------------------------------- VPN 核心操作 ------------------------------- def vpn_connected(vpn_name): """用 rasdial 判断连接状态（Get-VpnConnection 状态有缓存延迟，不可信）。""" out, _ = run_cmd(["rasdial.exe"]) return vpn_name in out def get_adapters(): """返回所有适配器（含隐藏）的字典列表。""" out, rc = run_ps( "Get-NetAdapter -IncludeHidden | " "Select-Object Name,ifIndex,Status,InterfaceDescription | ConvertTo-Json -Depth 3") if rc != 0 or not out.strip(): return [] try: data = json.loads(out) except json.JSONDecodeError: return [] if isinstance(data, dict): data = [data] return data def find_vpn_adapter(vpn_name, adapters=None): """ 定位 VPN 的 PPP 适配器。 注意：WAN Miniport (IP)/(IPv6)/(Network Monitor) 常驻 Up，不能用于匹配； 真正拨号成功时会出现 Name=连接名 的适配器（描述为 WAN Miniport (L2TP) 等）。 """ if adapters is None: adapters = get_adapters() for a in adapters: if a.get("Name") == vpn_name and a.get("Status") == "Up": return a for a in adapters: if (a.get("Status") == "Up" and re.search(r"WAN Miniport \((L2TP|PPTP|SSTP|IKEv2|PPPOE)\)", a.get("InterfaceDescription", ""))): return a # 兜底：部分 L2TP PPP 适配器在 Get-NetAdapter（即便 -IncludeHidden）里枚举不到， # 但会作为 IPv4 接口出现，别名 = 连接名。必须从 Get-NetIPAddress 反查。 out, rc = run_ps( "Get-NetIPAddress -AddressFamily IPv4 -ErrorAction SilentlyContinue | " "Where-Object {$_.InterfaceAlias -eq '%s' -and $_.IPAddress -ne '0.0.0.0'} | " "Select-Object -First 1 InterfaceAlias,ifIndex,IPAddress | ConvertTo-Json" % vpn_name) if rc == 0 and out.strip(): try: d = json.loads(out) if isinstance(d, dict) and d.get("ifIndex") is not None: return {"Name": d.get("InterfaceAlias"), "ifIndex": d["ifIndex"], "Status": "Up", "InterfaceDescription": "PPP over L2TP", "IPAddress": d.get("IPAddress")} except json.JSONDecodeError: pass return None def get_vpn_ip(if_index): out, _ = run_ps( "(Get-NetIPAddress -InterfaceIndex %s -AddressFamily IPv4 " "-ErrorAction SilentlyContinue | Where-Object {$_.IPAddress -ne '0.0.0.0'} " "| Select-Object -First 1).IPAddress" % if_index) ip = out.strip().splitlines() return ip[-1].strip() if ip and re.match(r"^\d+\.\d+\.\d+\.\d+$", ip[-1].strip()) else None def get_routes(prefix): out, _ = run_ps( "Get-NetRoute -AddressFamily IPv4 -ErrorAction SilentlyContinue | " "Where-Object {$_.DestinationPrefix -eq '%s'} | " "Select-Object ifIndex,NextHop,RouteMetric | ConvertTo-Json" % prefix) if not out.strip(): return [] try: data = json.loads(out) except json.JSONDecodeError: return [] return [data] if isinstance(data, dict) else data def get_default_routes(): return get_routes("0.0.0.0/0") def force_split_tunneling(vpn_name, is_all_user): """关闭远程默认网关（必须，否则 VPN 会抢占全部流量）。""" flag = " -AllUserConnection" if is_all_user else "" out, rc = run_ps( "$vc = Get-VpnConnection -Name '%s'%s -ErrorAction Stop; " "if ($vc.SplitTunneling -ne $true) { " " Set-VpnConnection -Name '%s'%s -SplitTunneling $true; 'CHANGED' " "} else { 'OK' }" % (vpn_name, flag, vpn_name, flag)) return out.strip(), rc def bind_connection_route(vpn_name, cidr, is_all_user): """把路由写进 VPN 连接（以后每次拨号自动生效）。""" flag = " -AllUserConnection" if is_all_user else "" run_ps("Remove-VpnConnectionRoute -ConnectionName '%s'%s -DestinationPrefix %s " "-ErrorAction SilentlyContinue" % (vpn_name, flag, cidr)) out, rc = run_ps("Add-VpnConnectionRoute -ConnectionName '%s'%s -DestinationPrefix %s " "-PassThru -ErrorAction Stop" % (vpn_name, flag, cidr)) return rc == 0 and cidr in out, out def dial(vpn_name, username="", password=""): if username and password: return run_cmd(["rasdial.exe", vpn_name, username, password]) return run_cmd(["rasdial.exe", vpn_name]) def disconnect(vpn_name): return run_cmd(["rasdial.exe", vpn_name, "/disconnect"]) def route_add(cidr, if_index, metric): net, mask = cidr.split("/") if int(mask) not in range(0, 33): return "掩码非法", -1 mask_str = mask_from_prefix(int(mask)) out, rc = run_cmd(["route.exe", "add", net, "mask", mask_str, "0.0.0.0", "IF", str(if_index), "metric", str(metric)]) return out, rc def mask_from_prefix(prefix_len): bits = "1" * prefix_len + "0" * (32 - prefix_len) return ".".join(str(int(bits[i:i + 8], 2)) for i in (0, 8, 16, 24)) def route_delete(cidr): net, mask = cidr.split("/") out, rc = run_cmd(["route.exe", "delete", net, "mask", mask_from_prefix(int(mask))]) return out, rc def disable_vpn_ipv6(adapter_name): return run_ps("Disable-NetAdapterBinding -Name '%s' -ComponentID ms_tcpip6 " "-ErrorAction Stop" % adapter_name) # --------------------------------- 业务流程 --------------------------------- def full_setup(cfg, log): """一键设置全流程。log(msg, level) 用于输出。""" vpn = cfg["vpn_name"] cidr = cfg["target_cidr"].strip() metric = int(cfg["route_metric"]) log("―――― 开始一键设置 ――――", "head") # 1. 连接存在性 out, rc = run_ps("Get-VpnConnection -Name '%s'%s" % (vpn, " -AllUserConnection" if cfg["is_all_user"] else "")) if rc != 0: log("[错误] 找不到 VPN 连接 [%s]，请检查连接名 / 是否全局连接" % vpn, "err") return False log("1/6 找到 VPN 连接", "ok") # 2. 分流模式 res, rc = force_split_tunneling(vpn, cfg["is_all_user"]) log("2/6 分流模式（本地流量走本地网关）：%s" % ("已是分流模式" if res == "OK" else ("已改设为分流" if res == "CHANGED" else res)), "ok") if rc != 0: log(" 设置分流失败：%s" % res, "err") # 3. 连接路由 ok, out = bind_connection_route(vpn, cidr, cfg["is_all_user"]) log("3/6 写入连接路由（重连自动生效）：%s" % ("成功" if ok else "失败：" + out.strip()), "ok" if ok else "warn") # 4. 拨号 if not vpn_connected(vpn): if cfg["auto_dial"]: log("4/6 正在拨号 [%s] ..." % vpn, "info") out, rc = dial(vpn, cfg["username"], cfg["password"]) time.sleep(1) if vpn_connected(vpn): log("4/6 拨号成功", "ok") else: log("4/6 拨号失败：%s" % out.strip().replace("\n", " "), "err") return False else: log("4/6 未自动拨号（请在系统里手动连接后点【补路由】）", "warn") return False else: log("4/6 VPN 已连接", "ok") # 5. 接口定位 ad = find_vpn_adapter(vpn) if not ad: # 适配器枚举不到，但连接路由可能已由 Windows 自动下发——此时分流其实已生效，不算失败 if get_routes(cidr): log("5/6 PPP 适配器未能枚举（L2TP 常见），但分流路由已由连接路由自动下发", "ok") log("―――― 设置完成（连接路由已生效）――――", "head") return True log("5/6 未检测到 VPN 虚拟网卡，且分流路由未生效（连接可能刚断开）", "err") return False log("5/6 VPN 接口：%s（ifIndex=%s）" % (ad["Name"], ad["ifIndex"]), "ok") if cfg["disable_vpn_ipv6"]: disable_vpn_ipv6(ad["Name"]) log(" 已关闭该适配器 IPv6", "ok") # 6. 路由 route_delete(cidr) out, rc = route_add(cidr, ad["ifIndex"], metric) if rc != 0 or not get_routes(cidr): ip = get_vpn_ip(ad["ifIndex"]) if ip: # 兜底：用 VPN 本机 IP 作下一跳再试 out, rc = run_cmd(["route.exe", "add"] + cidr.split("/")[0:1] + ["mask", mask_from_prefix(int(cidr.split("/")[1])), ip, "metric", str(metric)]) if get_routes(cidr): log("6/6 分流路由已生效：%s -> ifIndex %s" % (cidr, ad["ifIndex"]), "ok") log("―――― 设置完成 ――――", "head") return True log("6/6 路由添加失败：%s" % out.strip(), "err") return False def status_snapshot(cfg): """收集当前状态，供状态栏/测试模式使用。""" vpn = cfg["vpn_name"] cidr = cfg["target_cidr"].strip() connected = vpn_connected(vpn) # rasdial 为准（Get-VpnConnection 状态有缓存） ad = find_vpn_adapter(vpn) if connected else None routes = get_routes(cidr) defaults = get_default_routes() return { "connected": connected or ad is not None, "adapter": ad, "routes": routes, "default_gw": defaults[0]["NextHop"] if defaults else None, "default_if": defaults[0]["ifIndex"] if defaults else None, } # --------------------------------- GUI 界面 --------------------------------- class App: def __init__(self, root): self.root = root self.cfg = self.load_cfg() self.logq = queue.Queue() self.busy_lock = threading.Lock() self.keepalive_on = tk.BooleanVar(value=self.cfg["keepalive"]) root.title("VPN 分流助手 - %s" % self.cfg["vpn_name"]) root.geometry("760x640") self.build_ui() self.poll_log() self.refresh_status() self.set_keepalive(self.cfg["keepalive"]) # ---------- 配置 ---------- def load_cfg(self): cfg = dict(DEFAULTS) if os.path.exists(CONFIG_PATH): try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: cfg.update(json.load(f)) except Exception: pass return cfg def save_cfg(self): self.grab_vars() with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(self.cfg, f, ensure_ascii=False, indent=2) self.log("配置已保存到 " + CONFIG_PATH, "ok") # ---------- 界面 ---------- def build_ui(self): frm = ttk.LabelFrame(self.root, text="变量设置") frm.pack(fill="x", padx=10, pady=(10, 6)) self.v_vpn = tk.StringVar(value=self.cfg["vpn_name"]) self.v_cidr = tk.StringVar(value=self.cfg["target_cidr"]) self.v_metric = tk.StringVar(value=str(self.cfg["route_metric"])) self.v_user = tk.StringVar(value=self.cfg["username"]) self.v_pass = tk.StringVar(value=self.cfg["password"]) self.v_dial = tk.BooleanVar(value=self.cfg["auto_dial"]) self.v_ipv6 = tk.BooleanVar(value=self.cfg["disable_vpn_ipv6"]) self.v_alluser = tk.BooleanVar(value=self.cfg["is_all_user"]) rows = [("VPN 连接名", self.v_vpn, "entry"), ("目标网段 (CIDR)", self.v_cidr, "entry"), ("路由 metric", self.v_metric, "entry"), ("用户名（留空用已保存凭据）", self.v_user, "entry"), ("密码", self.v_pass, "pass")] for i, (label, var, kind) in enumerate(rows): ttk.Label(frm, text=label).grid(row=i // 2 * 2, column=(i % 2) * 2, sticky="e", padx=6, pady=3) w = ttk.Entry(frm, textvariable=var, width=28, show="*" if kind == "pass" else "") w.grid(row=i // 2 * 2, column=(i % 2) * 2 + 1, sticky="w", padx=4, pady=3) chk = ttk.Frame(frm); chk.grid(row=6, column=0, columnspan=4, sticky="w", padx=6) ttk.Checkbutton(chk, text="一键时自动拨号", variable=self.v_dial).pack(side="left", padx=4) ttk.Checkbutton(chk, text="断线自动重拨(保活)", variable=self.keepalive_on, command=lambda: self.set_keepalive(self.keepalive_on.get())).pack(side="left", padx=4) ttk.Checkbutton(chk, text="关闭VPN适配器IPv6", variable=self.v_ipv6).pack(side="left", padx=4) ttk.Checkbutton(chk, text="全局连接", variable=self.v_alluser).pack(side="left", padx=4) ttk.Button(frm, text="保存配置", command=self.save_cfg).grid( row=6, column=4, sticky="e", padx=6) # 按钮区 btns = ttk.Frame(self.root); btns.pack(fill="x", padx=10) ttk.Button(btns, text="⚡ 一键设置并连接", command=self.act_setup).pack(side="left", padx=3) ttk.Button(btns, text="断开 VPN", command=self.act_disconnect).pack(side="left", padx=3) ttk.Button(btns, text="移除分流路由", command=self.act_remove_route).pack(side="left", padx=3) ttk.Button(btns, text="刷新状态", command=self.refresh_status).pack(side="left", padx=3) # 状态栏 st = ttk.LabelFrame(self.root, text="当前状态") st.pack(fill="x", padx=10, pady=6) self.s_vpn = ttk.Label(st, text="VPN：-", font=("Consolas", 10)) self.s_route = ttk.Label(st, text="分流路由：-", font=("Consolas", 10)) self.s_gw = ttk.Label(st, text="默认网关：-", font=("Consolas", 10)) self.s_vpn.grid(row=0, column=0, sticky="w", padx=8, pady=3) self.s_route.grid(row=1, column=0, sticky="w", padx=8, pady=3) self.s_gw.grid(row=2, column=0, sticky="w", padx=8, pady=3) # 日志 lg = ttk.LabelFrame(self.root, text="运行日志") lg.pack(fill="both", expand=True, padx=10, pady=(0, 10)) self.txt = scrolledtext.ScrolledText(lg, height=14, font=("Consolas", 9), state="disabled") self.txt.pack(fill="both", expand=True, padx=4, pady=4) self.txt.tag_config("ok", foreground="#1a7f37") self.txt.tag_config("err", foreground="#c62828") self.txt.tag_config("warn", foreground="#b45309") self.txt.tag_config("info", foreground="#444") self.txt.tag_config("head", foreground="#0b5394", font=("Consolas", 9, "bold")) # ---------- 日志 ---------- def log(self, msg, level="info"): self.logq.put((msg, level)) def poll_log(self): try: while True: msg, lv = self.logq.get_nowait() self.txt.configure(state="normal") self.txt.insert("end", time.strftime("[%H:%M:%S] ") + msg + "\n", lv) self.txt.see("end") self.txt.configure(state="disabled") except queue.Empty: pass self.root.after(200, self.poll_log) # ---------- 变量 ---------- def grab_vars(self): self.cfg.update({ "vpn_name": self.v_vpn.get().strip(), "target_cidr": self.v_cidr.get().strip(), "route_metric": int(self.v_metric.get() or 1), "username": self.v_user.get().strip(), "password": self.v_pass.get(), "auto_dial": self.v_dial.get(), "keepalive": self.keepalive_on.get(), "disable_vpn_ipv6": self.v_ipv6.get(), "is_all_user": self.v_alluser.get(), }) # ---------- 动作 ---------- def act_setup(self): self.grab_vars() if not re.match(r"^\d+\.\d+\.\d+\.\d+/\d{1,2}$", self.cfg["target_cidr"]): messagebox.showerror("格式错误", "目标网段需为 CIDR 格式，如 58.220.83.0/24") return threading.Thread(target=self._do_setup, daemon=True).start() def _do_setup(self): if not self.busy_lock.acquire(blocking=False): self.log("已有任务在执行，请稍候", "warn"); return try: full_setup(self.cfg, self.log) self.refresh_status() finally: self.busy_lock.release() def act_disconnect(self): self.grab_vars() out, rc = disconnect(self.cfg["vpn_name"]) self.log("断开 VPN：%s" % out.strip(), "info" if rc == 0 else "err") self.refresh_status() def act_remove_route(self): self.grab_vars() out, rc = route_delete(self.cfg["target_cidr"]) self.log("移除分流路由：%s" % out.strip(), "info") self.refresh_status() def refresh_status(self): def worker(): snap = status_snapshot(self.cfg) if snap["connected"]: a = snap["adapter"] self.s_vpn.config(text="VPN：已连接 %s（ifIndex=%s，IP %s）" % ( a["Name"], a["ifIndex"], get_vpn_ip(a["ifIndex"]) or "?")) else: self.s_vpn.config(text="VPN：未连接") if snap["routes"]: r = snap["routes"][0] self.s_route.config(text="分流路由：已生效 -> ifIndex %s（metric %s）" % ( r["ifIndex"], r["RouteMetric"])) else: self.s_route.config(text="分流路由：未生效") self.s_gw.config(text="默认网关：%s（ifIndex %s）← 本地流量走这里" % ( snap["default_gw"], snap["default_if"])) threading.Thread(target=worker, daemon=True).start() # ---------- 保活 ---------- def set_keepalive(self, enabled): self.keepalive_on.set(enabled) if enabled: threading.Thread(target=self._keepalive_loop, daemon=True).start() self.log("保活已开启：每 %ds 检查一次，断线自动重拨补路由" % KEEPALIVE_INTERVAL, "info") def _keepalive_loop(self): while self.keepalive_on.get(): time.sleep(KEEPALIVE_INTERVAL) if not self.keepalive_on.get() or self.busy_lock.locked(): continue vpn = self.cfg["vpn_name"] cidr = self.cfg["target_cidr"] if not vpn_connected(vpn): self.log("[保活] 检测到 VPN 断开，自动重拨...", "warn") with self.busy_lock: full_setup(self.cfg, self.log) self.refresh_status() elif not get_routes(cidr): ad = find_vpn_adapter(vpn) if ad: self.log("[保活] 路由丢失，自动补加...", "warn") route_delete(cidr) route_add(cidr, ad["ifIndex"], int(self.cfg["route_metric"])) self.refresh_status() # --------------------------------- 入口 --------------------------------- def selftest(): """--test：不开窗口，跑核心检测验证环境。""" print("IsAdmin:", is_admin()) print("rasdial output:", run_cmd(["rasdial.exe"])[0].strip()) print("connected:", vpn_connected(DEFAULTS["vpn_name"])) ads = get_adapters() print("adapters:", len(ads)) for a in ads: print(" ", a) ad = find_vpn_adapter(DEFAULTS["vpn_name"], ads) print("vpn adapter:", ad) print("mask(/24):", mask_from_prefix(24)) print("routes:", get_routes(DEFAULTS["target_cidr"])) print("default:", get_default_routes()) def main(): if "--test" in sys.argv: selftest(); return if os.name != "nt": print("仅支持 Windows"); return if not is_admin(): elevate(); return root = tk.Tk() try: ctypes.windll.shcore.SetProcessDpiAwareness(1) except Exception: pass App(root) root.mainloop() if __name__ == "__main__": main()