fix(lite): hide tun network command windows
This commit is contained in:
@@ -27,6 +27,16 @@ class TunBackendError(RuntimeError):
|
||||
"""TUN 后端无法启动或停止。"""
|
||||
|
||||
|
||||
def _run_hidden(command, **kwargs):
|
||||
"""运行 Windows 网络命令时隐藏控制台窗口。
|
||||
|
||||
源码通过 test.bat 启动时这些子进程复用已有控制台;Nuitka GUI 没有父控制台,
|
||||
未设置 CREATE_NO_WINDOW 就会为每次 PowerShell 查询弹出一个窗口。
|
||||
"""
|
||||
kwargs.setdefault("creationflags", getattr(subprocess, "CREATE_NO_WINDOW", 0))
|
||||
return subprocess.run(command, **kwargs)
|
||||
|
||||
|
||||
class TunBackend:
|
||||
def __init__(self, user_dir: str | Path, engine_path: str | Path | None = None,
|
||||
log: Callable[[str], None] | None = None,
|
||||
@@ -110,7 +120,7 @@ class TunBackend:
|
||||
"Where-Object InterfaceAlias -ne 'RRPP-TUN' | "
|
||||
"Sort-Object RouteMetric | Select-Object -First 1 -ExpandProperty InterfaceAlias)"]
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=5)
|
||||
result = _run_hidden(command, capture_output=True, text=True, timeout=5)
|
||||
value = (result.stdout or "").strip().splitlines()
|
||||
if value and value[0].strip():
|
||||
return value[0].strip()
|
||||
@@ -127,7 +137,7 @@ class TunBackend:
|
||||
f"(Get-NetIPAddress -InterfaceAlias '{escaped}' -AddressFamily IPv4 "
|
||||
"-AddressState Preferred | Select-Object -First 1 -ExpandProperty IPAddress)"]
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=5)
|
||||
result = _run_hidden(command, capture_output=True, text=True, timeout=5)
|
||||
value = (result.stdout or "").strip().splitlines()
|
||||
if value and value[0].strip():
|
||||
return value[0].strip()
|
||||
@@ -144,7 +154,7 @@ class TunBackend:
|
||||
"Sort-Object RouteMetric | Select-Object -First 1 "
|
||||
"NextHop,InterfaceIndex | ConvertTo-Json -Compress"]
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=5)
|
||||
result = _run_hidden(command, capture_output=True, text=True, timeout=5)
|
||||
data = json.loads((result.stdout or "").strip())
|
||||
if isinstance(data, dict):
|
||||
return str(data.get("NextHop") or ""), str(data.get("InterfaceIndex") or "")
|
||||
@@ -161,7 +171,7 @@ class TunBackend:
|
||||
"Sort-Object RouteMetric | Select-Object -First 1 "
|
||||
"NextHop,InterfaceIndex | ConvertTo-Json -Compress"]
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=5)
|
||||
result = _run_hidden(command, capture_output=True, text=True, timeout=5)
|
||||
data = json.loads((result.stdout or "").strip())
|
||||
if isinstance(data, dict):
|
||||
return str(data.get("NextHop") or ""), str(data.get("InterfaceIndex") or "")
|
||||
@@ -182,7 +192,7 @@ class TunBackend:
|
||||
"$_.IPAddress -notlike 'fd*' }).IPAddress | ConvertTo-Json -Compress",
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, timeout=5)
|
||||
result = _run_hidden(command, capture_output=True, text=True, timeout=5)
|
||||
if result.returncode:
|
||||
return None
|
||||
value = json.loads((result.stdout or "").strip() or "[]")
|
||||
@@ -242,7 +252,7 @@ class TunBackend:
|
||||
if self._physical_gateway:
|
||||
command.append(self._physical_gateway)
|
||||
command.extend(["METRIC", "1", "IF", self._physical_ifindex])
|
||||
result = subprocess.run(command, capture_output=True, text=True,
|
||||
result = _run_hidden(command, capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
if result.returncode:
|
||||
detail = (result.stderr or result.stdout or "").strip()
|
||||
@@ -275,7 +285,7 @@ class TunBackend:
|
||||
if self._physical_ipv6_gateway and self._physical_ipv6_gateway != "::":
|
||||
command.append(f"nexthop={self._physical_ipv6_gateway}")
|
||||
command.append("metric=1")
|
||||
result = subprocess.run(["netsh", *command], capture_output=True, text=True,
|
||||
result = _run_hidden(["netsh", *command], capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
if result.returncode:
|
||||
detail = (result.stderr or result.stdout or "").strip()
|
||||
@@ -310,7 +320,7 @@ class TunBackend:
|
||||
if self._physical_gateway:
|
||||
command.append(self._physical_gateway)
|
||||
command.extend(["METRIC", "1", "IF", self._physical_ifindex])
|
||||
result = subprocess.run(command, capture_output=True, text=True,
|
||||
result = _run_hidden(command, capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
detail = (result.stderr or result.stdout or "").strip()
|
||||
if result.returncode:
|
||||
@@ -333,7 +343,7 @@ class TunBackend:
|
||||
|
||||
def _cleanup_netsh(self, args: list[str]) -> bool:
|
||||
try:
|
||||
result = subprocess.run(["netsh", *args], capture_output=True, text=True,
|
||||
result = _run_hidden(["netsh", *args], capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
except Exception as exc:
|
||||
self.detail_log(f"[TUN] 清理 netsh 失败: {exc}")
|
||||
@@ -348,7 +358,7 @@ class TunBackend:
|
||||
remaining: list[str] = []
|
||||
for address in self._node_routes:
|
||||
try:
|
||||
result = subprocess.run(["route", "DELETE", address], capture_output=True,
|
||||
result = _run_hidden(["route", "DELETE", address], capture_output=True,
|
||||
text=True, encoding="mbcs", errors="replace", timeout=15)
|
||||
except Exception as exc:
|
||||
self.detail_log(f"[TUN] 清理节点路由失败 ({address}): {exc}")
|
||||
@@ -365,7 +375,7 @@ class TunBackend:
|
||||
if self._physical_ipv6_ifindex:
|
||||
command.append(f"interface={self._physical_ipv6_ifindex}")
|
||||
try:
|
||||
result = subprocess.run(["netsh", *command], capture_output=True, text=True,
|
||||
result = _run_hidden(["netsh", *command], capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
except Exception as exc:
|
||||
self.detail_log(f"[TUN] 清理 IPv6 节点路由失败 ({address}): {exc}")
|
||||
@@ -379,7 +389,7 @@ class TunBackend:
|
||||
return not remaining and not remaining_v6
|
||||
|
||||
def _netsh(self, args: list[str], *, ignore_error: bool = False) -> None:
|
||||
result = subprocess.run(["netsh", *args], capture_output=True, text=True,
|
||||
result = _run_hidden(["netsh", *args], capture_output=True, text=True,
|
||||
encoding="mbcs", errors="replace", timeout=15)
|
||||
if result.returncode and not ignore_error:
|
||||
detail = (result.stderr or result.stdout or "").strip()
|
||||
@@ -388,7 +398,7 @@ class TunBackend:
|
||||
def _configure_routes(self) -> None:
|
||||
for _ in range(20):
|
||||
try:
|
||||
probe = subprocess.run(
|
||||
probe = _run_hidden(
|
||||
["powershell", "-NoProfile", "-Command",
|
||||
f"Get-NetAdapter -Name '{self._device}' -ErrorAction SilentlyContinue"],
|
||||
capture_output=True, text=True, timeout=3,
|
||||
|
||||
Reference in New Issue
Block a user