loop-uh
d5a4715771
Some checks failed
Windows project source guards / test (push) Has been cancelled
550 lines
21 KiB
Python
550 lines
21 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import deque
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import time
|
||
from typing import Any
|
||
|
||
_CREATE_NO_WINDOW = 0x08000000 if os.name == "nt" else 0
|
||
|
||
from PyQt6.QtCore import QObject, QProcess, pyqtSignal
|
||
|
||
from ...platform.windows import win_netinfo
|
||
from ...diagnostics.export import capture_runtime_config
|
||
from ...constants import RUNTIME_DIR, SINGBOX_CONFIG_FILE, SINGBOX_PATH_DEFAULT
|
||
from ...profiles.path_utils import resolve_configured_path
|
||
from ...network.proxy_readiness import probe_listener_role
|
||
from .config_check import check_config
|
||
from ...platform.windows.subprocess_utils import (
|
||
decode_output,
|
||
kill_processes_by_path,
|
||
result_output_text,
|
||
run_text_pumped,
|
||
sleep_with_events,
|
||
wait_for_qprocess_finished,
|
||
wait_for_qprocess_started,
|
||
)
|
||
|
||
|
||
class SingBoxManager(QObject):
|
||
started = pyqtSignal()
|
||
stopped = pyqtSignal(int)
|
||
log_received = pyqtSignal(str)
|
||
error = pyqtSignal(str)
|
||
state_changed = pyqtSignal(bool)
|
||
|
||
def __init__(self, parent: QObject | None = None):
|
||
super().__init__(parent)
|
||
self._process = QProcess(self)
|
||
self._process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
|
||
self._process.readyReadStandardOutput.connect(self._on_ready_read)
|
||
self._process.started.connect(self._on_started)
|
||
self._process.errorOccurred.connect(self._on_error)
|
||
self._process.finished.connect(self._on_finished)
|
||
self._running = False
|
||
self._starting = False
|
||
self._stop_requested = False
|
||
self._startup_failure_reported = False
|
||
self._runtime_error_reported = False
|
||
self._last_output_lines: deque[str] = deque(maxlen=20)
|
||
self._last_exit_code: int | None = None
|
||
self._last_exit_status = QProcess.ExitStatus.NormalExit
|
||
self._uses_tun = False
|
||
self._last_start_failure_retryable = False
|
||
self.diagnostic_config: dict[str, Any] | None = None
|
||
|
||
@property
|
||
def is_running(self) -> bool:
|
||
return self._running
|
||
|
||
@property
|
||
def last_start_failure_retryable(self) -> bool:
|
||
return self._last_start_failure_retryable
|
||
|
||
def start(self, singbox_path: str, config: dict[str, Any]) -> bool:
|
||
self._last_start_failure_retryable = False
|
||
exe = resolve_configured_path(
|
||
singbox_path,
|
||
default_path=SINGBOX_PATH_DEFAULT,
|
||
use_default_if_empty=True,
|
||
migrate_default_location=True,
|
||
)
|
||
if exe is None:
|
||
self.error.emit("sing-box path is not configured (set it in Settings → Core paths)")
|
||
return False
|
||
if not exe.is_file():
|
||
self.error.emit(f"sing-box.exe not found: {exe}")
|
||
return False
|
||
|
||
tun_interface_name = self._extract_tun_interface_name(config)
|
||
uses_tun = bool(tun_interface_name)
|
||
|
||
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
||
SINGBOX_CONFIG_FILE.write_text(
|
||
json.dumps(config, ensure_ascii=True, indent=2), encoding="utf-8"
|
||
)
|
||
self.diagnostic_config = capture_runtime_config(exe, config)
|
||
|
||
# Ядро отвергает негодную конфигурацию уже после старта процесса, печатая
|
||
# её причину в свой лог вперемешку с ANSI-кодами. Спросим его заранее,
|
||
# чтобы пользователь увидел, какое именно поле не принято.
|
||
config_ok, config_problem = check_config(exe, SINGBOX_CONFIG_FILE)
|
||
if not config_ok:
|
||
self.error.emit(config_problem)
|
||
return False
|
||
if config_problem:
|
||
self.log_received.emit(f"[singbox] {config_problem}")
|
||
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
if not self.stop(expected=True):
|
||
self.error.emit("failed to stop previous sing-box process")
|
||
return False
|
||
elif self._running:
|
||
self._running = False
|
||
self.state_changed.emit(False)
|
||
|
||
# Kill an orphaned process before reusing its ports or TUN adapter.
|
||
self._kill_orphaned(exe)
|
||
self._uses_tun = uses_tun
|
||
|
||
# Set working directory to core/ so sing-box can find wintun.dll
|
||
core_dir = exe.parent
|
||
self._starting = True
|
||
self._startup_failure_reported = False
|
||
self._runtime_error_reported = False
|
||
self._last_output_lines.clear()
|
||
|
||
# Proxy mode has no adapter to wait for. Its Clash API is the runtime
|
||
# control plane, so process existence alone is not readiness: local
|
||
# rule-set loading may still be in progress after QProcess.started.
|
||
if not tun_interface_name:
|
||
self._last_output_lines.clear()
|
||
self._process.setWorkingDirectory(str(core_dir))
|
||
self._process.setProgram(str(exe))
|
||
self._process.setArguments(["run", "-c", str(SINGBOX_CONFIG_FILE), "-D", str(core_dir)])
|
||
self._process.start()
|
||
if not wait_for_qprocess_started(self._process, 4000):
|
||
self._starting = False
|
||
self._report_startup_failure(f"failed to start sing-box process: {self._process.errorString()}")
|
||
return False
|
||
if not self._wait_until_proxy_ready(config):
|
||
self._starting = False
|
||
return False
|
||
self._starting = False
|
||
self._mark_running()
|
||
return True
|
||
|
||
# TUN mode retries while Windows releases a previous wintun adapter.
|
||
for attempt in range(3):
|
||
self._last_output_lines.clear()
|
||
self._process.setWorkingDirectory(str(core_dir))
|
||
self._process.setProgram(str(exe))
|
||
self._process.setArguments(["run", "-c", str(SINGBOX_CONFIG_FILE), "-D", str(core_dir)])
|
||
self._process.start()
|
||
|
||
if not wait_for_qprocess_started(self._process, 4000):
|
||
self._starting = False
|
||
self._report_startup_failure(f"failed to start sing-box process: {self._process.errorString()}")
|
||
return False
|
||
|
||
if self._wait_until_tun_ready(tun_interface_name):
|
||
self._starting = False
|
||
self._mark_running()
|
||
return True
|
||
|
||
exited = self._process.state() == QProcess.ProcessState.NotRunning
|
||
retryable = exited and self._startup_error_is_retryable()
|
||
if not exited:
|
||
self.stop(expected=True)
|
||
|
||
if retryable and attempt < 2:
|
||
self._wait_tun_released()
|
||
self._starting = True
|
||
continue
|
||
|
||
self._starting = False
|
||
if exited:
|
||
self._report_startup_failure(
|
||
self._unexpected_exit_message(self._last_exit_code, self._last_exit_status, startup=True)
|
||
)
|
||
else:
|
||
self._report_startup_failure(
|
||
f"sing-box started but TUN interface '{tun_interface_name}' did not become ready in time"
|
||
)
|
||
return False
|
||
|
||
self._starting = False
|
||
return False
|
||
|
||
@staticmethod
|
||
def _kill_orphaned(exe: Path) -> None:
|
||
"""Kill orphaned sing-box processes that hold the TUN adapter."""
|
||
if os.name != "nt":
|
||
return
|
||
try:
|
||
if kill_processes_by_path(exe.name, exe, timeout=5):
|
||
sleep_with_events(1.0)
|
||
except Exception:
|
||
pass
|
||
|
||
def stop(self, expected: bool = True) -> bool:
|
||
if self._process.state() == QProcess.ProcessState.NotRunning:
|
||
self._stop_requested = False
|
||
if self._running:
|
||
self._running = False
|
||
self.state_changed.emit(False)
|
||
self._starting = False
|
||
self._uses_tun = False
|
||
return True
|
||
|
||
used_tun = self._uses_tun
|
||
self._stop_requested = expected
|
||
# Короткий grace на снятие TUN-адаптера: на Windows консольный sing-box
|
||
# игнорирует terminate() (WM_CLOSE), поэтому ждём не дольше 500мс и убиваем.
|
||
self._process.terminate()
|
||
if not wait_for_qprocess_finished(self._process, 500):
|
||
self._process.kill()
|
||
wait_for_qprocess_finished(self._process, 2000)
|
||
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
self._stop_requested = False
|
||
self.error.emit("failed to stop sing-box process in time")
|
||
return False
|
||
|
||
self._uses_tun = False
|
||
self._starting = False
|
||
if used_tun:
|
||
self._wait_tun_released()
|
||
return True
|
||
|
||
@staticmethod
|
||
def _wait_tun_released(max_wait: float = 10.0) -> None:
|
||
"""Poll until the TUN adapter is gone, up to max_wait seconds."""
|
||
if os.name != "nt":
|
||
return
|
||
waited = 0.0
|
||
while waited < max_wait:
|
||
gone, fast_probe = SingBoxManager._probe_tun_adapter_gone()
|
||
if gone:
|
||
return
|
||
step = 0.1 if fast_probe else 0.3
|
||
sleep_with_events(step)
|
||
waited += step
|
||
|
||
@staticmethod
|
||
def _probe_tun_adapter_gone() -> tuple[bool, bool]:
|
||
"""Return (adapter is gone, fast ctypes path was used)."""
|
||
if win_netinfo.is_available():
|
||
try:
|
||
return (not win_netinfo.any_adapter_name_contains("xftun")), True
|
||
except Exception:
|
||
pass # fall back to netsh below
|
||
try:
|
||
result = run_text_pumped(
|
||
["netsh", "interface", "show", "interface"],
|
||
timeout=3,
|
||
creationflags=_CREATE_NO_WINDOW,
|
||
)
|
||
# Check if any xftun* adapter still exists
|
||
return ("xftun" not in result_output_text(result)), False
|
||
except Exception:
|
||
return True, False # can't check, proceed anyway
|
||
|
||
def _on_ready_read(self) -> None:
|
||
chunk = self._process.readAllStandardOutput()
|
||
raw = getattr(chunk, "data")()
|
||
if isinstance(raw, (bytes, bytearray)):
|
||
text = decode_output(bytes(raw))
|
||
else:
|
||
text = str(raw)
|
||
for line in text.splitlines():
|
||
clean = line.rstrip()
|
||
if clean:
|
||
self._last_output_lines.append(clean)
|
||
self.log_received.emit(clean)
|
||
|
||
def _on_started(self) -> None:
|
||
self._stop_requested = False
|
||
|
||
def _on_error(self, process_error: QProcess.ProcessError) -> None:
|
||
if self._stop_requested and process_error == QProcess.ProcessError.Crashed:
|
||
return
|
||
message = f"sing-box process error: {process_error.name} ({self._process.errorString()})"
|
||
if self._starting:
|
||
self._report_startup_failure(message)
|
||
return
|
||
if self._runtime_error_reported:
|
||
return
|
||
self._runtime_error_reported = True
|
||
self.error.emit(message)
|
||
|
||
def _on_finished(self, exit_code: int, _exit_status: int = 0) -> None:
|
||
exit_status = QProcess.ExitStatus(_exit_status)
|
||
expected = self._stop_requested
|
||
self._last_exit_code = exit_code
|
||
self._last_exit_status = exit_status
|
||
self._stop_requested = False
|
||
was_running = self._running
|
||
self._running = False
|
||
self._uses_tun = False
|
||
if self._starting and not expected:
|
||
self._report_startup_failure(self._unexpected_exit_message(exit_code, exit_status, startup=True))
|
||
elif was_running and not expected and not self._runtime_error_reported:
|
||
self._runtime_error_reported = True
|
||
self.error.emit(self._unexpected_exit_message(exit_code, exit_status, startup=False))
|
||
self._starting = False
|
||
self.stopped.emit(exit_code)
|
||
if was_running:
|
||
self.state_changed.emit(False)
|
||
|
||
def _mark_running(self) -> None:
|
||
if self._running:
|
||
return
|
||
self._stop_requested = False
|
||
self._running = True
|
||
self.started.emit()
|
||
self.state_changed.emit(True)
|
||
|
||
@staticmethod
|
||
def _extract_tun_interface_name(config: dict[str, Any]) -> str:
|
||
for inbound in config.get("inbounds") or []:
|
||
if not isinstance(inbound, dict):
|
||
continue
|
||
if str(inbound.get("type") or "").strip().lower() != "tun":
|
||
continue
|
||
return str(inbound.get("interface_name") or "").strip()
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _extract_clash_api_port(config: dict[str, Any]) -> int:
|
||
experimental = config.get("experimental")
|
||
if not isinstance(experimental, dict):
|
||
return 0
|
||
clash_api = experimental.get("clash_api")
|
||
if not isinstance(clash_api, dict):
|
||
return 0
|
||
endpoint = str(clash_api.get("external_controller") or "").strip()
|
||
if ":" not in endpoint:
|
||
return 0
|
||
try:
|
||
port = int(endpoint.rsplit(":", 1)[1])
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
return port if 0 < port <= 65535 else 0
|
||
|
||
@classmethod
|
||
def _extract_proxy_port_roles(cls, config: dict[str, Any]) -> dict[int, str]:
|
||
roles: dict[int, str] = {}
|
||
for inbound in config.get("inbounds") or []:
|
||
if not isinstance(inbound, dict):
|
||
continue
|
||
inbound_type = str(inbound.get("type") or "").strip().lower()
|
||
if inbound_type not in {"socks", "mixed", "http"}:
|
||
continue
|
||
try:
|
||
port = int(inbound.get("listen_port") or 0)
|
||
except (TypeError, ValueError):
|
||
port = 0
|
||
if port > 0:
|
||
roles[port] = "HTTP" if inbound_type == "http" else "SOCKS"
|
||
api_port = cls._extract_clash_api_port(config)
|
||
if api_port > 0:
|
||
roles[api_port] = "Clash API"
|
||
return roles
|
||
|
||
@staticmethod
|
||
def _extract_socks_credentials(config: dict[str, Any]) -> dict[int, dict[str, str]]:
|
||
credentials: dict[int, dict[str, str]] = {}
|
||
for inbound in config.get("inbounds") or []:
|
||
if not isinstance(inbound, dict):
|
||
continue
|
||
if str(inbound.get("type") or "").strip().lower() not in {"socks", "mixed"}:
|
||
continue
|
||
try:
|
||
port = int(inbound.get("listen_port") or 0)
|
||
except (TypeError, ValueError):
|
||
port = 0
|
||
if port <= 0:
|
||
continue
|
||
users = inbound.get("users")
|
||
if not isinstance(users, list) or not users:
|
||
continue
|
||
first = users[0]
|
||
if not isinstance(first, dict):
|
||
continue
|
||
username = str(first.get("username") or "")
|
||
password = str(first.get("password") or "")
|
||
if username:
|
||
credentials[port] = {"username": username, "password": password}
|
||
return credentials
|
||
|
||
def _wait_until_proxy_ready(
|
||
self,
|
||
config: dict[str, Any],
|
||
max_wait: float = 15.0,
|
||
) -> bool:
|
||
port_roles = self._extract_proxy_port_roles(config)
|
||
port_credentials = self._extract_socks_credentials(config)
|
||
if not port_roles:
|
||
sleep_with_events(0.75)
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
return True
|
||
self._report_startup_failure(
|
||
self._unexpected_exit_message(
|
||
self._last_exit_code,
|
||
self._last_exit_status,
|
||
startup=True,
|
||
)
|
||
)
|
||
return False
|
||
|
||
deadline = time.monotonic() + max(0.0, max_wait)
|
||
while time.monotonic() < deadline:
|
||
if self._process.state() == QProcess.ProcessState.NotRunning:
|
||
self._report_startup_failure(
|
||
self._unexpected_exit_message(
|
||
self._last_exit_code,
|
||
self._last_exit_status,
|
||
startup=True,
|
||
)
|
||
)
|
||
return False
|
||
if all(
|
||
probe_listener_role(port, role, **(port_credentials.get(port) or {}))
|
||
for port, role in port_roles.items()
|
||
):
|
||
return True
|
||
sleep_with_events(0.1)
|
||
|
||
self._last_start_failure_retryable = True
|
||
pending = [
|
||
f"{role} {port}"
|
||
for port, role in port_roles.items()
|
||
if not probe_listener_role(port, role, **(port_credentials.get(port) or {}))
|
||
]
|
||
# Диагностику ядра нужно снять до stop(): штатная остановка стирает
|
||
# контекст последних строк как «ожидаемый» выход.
|
||
last_output = next(
|
||
(line for line in reversed(self._last_output_lines) if line.strip()), ""
|
||
)
|
||
self.stop(expected=True)
|
||
message = (
|
||
"sing-box запустился, но локальные входы не подтвердили готовность: "
|
||
+ (", ".join(pending) if pending else "неизвестный вход")
|
||
)
|
||
if last_output:
|
||
message = f"{message}. Последний вывод ядра: {last_output}"
|
||
self._report_startup_failure(message)
|
||
return False
|
||
|
||
def _wait_until_tun_ready(self, tun_interface_name: str, max_wait: float = 18.0) -> bool:
|
||
if os.name != "nt" or not tun_interface_name:
|
||
return True
|
||
deadline = time.monotonic() + max(0.0, max_wait)
|
||
while time.monotonic() < deadline:
|
||
if self._process.state() == QProcess.ProcessState.NotRunning:
|
||
return False
|
||
has_ipv4, fast_probe = self._probe_tun_interface_has_ipv4(tun_interface_name)
|
||
if has_ipv4:
|
||
return True
|
||
step = 0.1 if fast_probe else 0.25
|
||
sleep_with_events(step)
|
||
return False
|
||
|
||
@classmethod
|
||
def _probe_tun_interface_has_ipv4(cls, tun_interface_name: str) -> tuple[bool, bool]:
|
||
"""Return (interface has an IPv4 address, fast ctypes path was used)."""
|
||
if win_netinfo.is_available():
|
||
try:
|
||
if win_netinfo.adapter_has_ipv4(tun_interface_name):
|
||
return True, True
|
||
except Exception:
|
||
pass
|
||
|
||
# A negative GetAdaptersAddresses result is not authoritative for
|
||
# a newly-created Wintun adapter: Windows may expose it to the
|
||
# NetTCPIP provider before it is visible in the ordinary adapter
|
||
# list. Verify the negative result through the independent legacy
|
||
# provider instead of timing out and killing a healthy sing-box.
|
||
return cls._tun_interface_has_ipv4(tun_interface_name), False
|
||
return cls._tun_interface_has_ipv4(tun_interface_name), False
|
||
|
||
@staticmethod
|
||
def _tun_interface_has_ipv4(tun_interface_name: str) -> bool:
|
||
"""Legacy PowerShell probe, kept as a fallback for the ctypes fast path."""
|
||
escaped_name = tun_interface_name.replace("'", "''")
|
||
script = (
|
||
f"$ipv4 = Get-NetIPAddress -InterfaceAlias '{escaped_name}' -AddressFamily IPv4 -ErrorAction SilentlyContinue "
|
||
"| Where-Object { $_.IPAddress -and $_.IPAddress -ne '0.0.0.0' } "
|
||
"| Select-Object -First 1 IPAddress; "
|
||
"if ($ipv4) { exit 0 } else { exit 1 }"
|
||
)
|
||
try:
|
||
result = run_text_pumped(
|
||
["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
|
||
timeout=4,
|
||
check=False,
|
||
creationflags=_CREATE_NO_WINDOW,
|
||
)
|
||
except Exception:
|
||
return False
|
||
return result.returncode == 0
|
||
|
||
def _startup_error_is_retryable(self) -> bool:
|
||
needles = ("already exists", "cannot create a file when that file already exists")
|
||
for line in self._last_output_lines:
|
||
text = line.lower()
|
||
if any(needle in text for needle in needles):
|
||
return True
|
||
return False
|
||
|
||
def _unexpected_exit_message(
|
||
self,
|
||
exit_code: int | None,
|
||
exit_status: QProcess.ExitStatus,
|
||
*,
|
||
startup: bool,
|
||
) -> str:
|
||
stage = "during startup" if startup else "unexpectedly"
|
||
detail = self._last_output_lines[-1].strip() if self._last_output_lines else ""
|
||
if detail:
|
||
return f"sing-box exited {stage}: {detail}"
|
||
if exit_code is None:
|
||
return f"sing-box exited {stage}."
|
||
status_name = "CrashExit" if exit_status == QProcess.ExitStatus.CrashExit else "NormalExit"
|
||
return f"sing-box exited {stage} with code {exit_code} ({status_name})."
|
||
|
||
def _report_startup_failure(self, message: str) -> None:
|
||
if self._startup_failure_reported:
|
||
return
|
||
self._startup_failure_reported = True
|
||
self.error.emit(message)
|
||
|
||
|
||
def get_singbox_version(singbox_path: str) -> str | None:
|
||
exe = resolve_configured_path(
|
||
singbox_path,
|
||
default_path=SINGBOX_PATH_DEFAULT,
|
||
use_default_if_empty=True,
|
||
migrate_default_location=True,
|
||
)
|
||
if exe is None:
|
||
return None
|
||
if not exe.exists():
|
||
return None
|
||
try:
|
||
result = run_text_pumped(
|
||
[str(exe), "version"],
|
||
timeout=3,
|
||
check=False,
|
||
creationflags=_CREATE_NO_WINDOW,
|
||
)
|
||
except Exception:
|
||
return None
|
||
|
||
lines = result_output_text(result).splitlines()
|
||
if not lines:
|
||
return None
|
||
return lines[0].strip()
|