loop-uh
e26419216a
Some checks failed
Windows project source guards / test (push) Failing after 1m12s
Пользователь 0.8.36: VK и Яндекс перестали открываться, пока не добавил
прямые правила. В диагностике через туннель шли 29–39 % соединений к
российским IP (geoip-ru), но по ней нельзя было понять, каким правилом:
сайдкар видит только IP, а конфиг — уже исправленный. Шаблон и данные
маршрутизации в 0.8.30…0.8.36 не менялись (srs побайтно равны); причина —
правило пользователя, вставшее выше стокового «РФ напрямую».
- «Простое правило»: программа «Через VPN» по умолчанию «Кроме российских
сайтов» — native logical and с {rule_set geosite-category-ru/geoip-ru (если
есть в документе), domain_suffix ru/xn--p1ai/su, invert}; российское
назначение идёт дальше по списку. Без галочки — предупреждение, что VK,
Яндекс, банки уйдут в туннель. Семантика проверена ядром той же версии:
`domain_suffix && !(domain_suffix=[ru xn--p1ai su]) => route(...)`.
- diagnostics/route_decisions: сводка решений маршрута по последним четырём
сессиям (процесс, сайт из двух меток или IP, правило ядра «… => …»/final,
маршрут, число соединений; отдельно российские сайты в туннеле) из того же
опроса Clash API /connections, что метрики; журнал живёт у контроллера и
переживает переподключение.
- runtime_redacted: route_history — правила route последних восьми запусков
sing-box.
- Тесты: форма правила, explain_route (российские сайты мимо правила к
стоковому direct), страница (галочка, предупреждение, «Напрямую»), сводка и
история; docs/sing-box/routing-gui.md.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
715 lines
30 KiB
Python
715 lines
30 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import deque
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import time
|
||
from typing import Any, Callable
|
||
|
||
_CREATE_NO_WINDOW = 0x08000000 if os.name == "nt" else 0
|
||
|
||
from PyQt6.QtCore import QObject, QProcess, pyqtSignal
|
||
|
||
from ...application.async_steps import (
|
||
TransitionSteps,
|
||
run_in_worker,
|
||
run_steps_blocking,
|
||
sleep_ms,
|
||
wait_process_finished,
|
||
wait_process_started,
|
||
)
|
||
from ...platform.windows import win_netinfo
|
||
from ...diagnostics.export import capture_runtime_config, route_history_entry
|
||
from ...constants import RUNTIME_DIR, SINGBOX_CONFIG_FILE, SINGBOX_PATH_DEFAULT
|
||
from ...profiles.path_utils import resolve_configured_path
|
||
from ...network.proxy_readiness import LISTENER_POLL_MS, ports_listening, 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,
|
||
run_text_pumped,
|
||
)
|
||
|
||
|
||
def _still_wanted(should_continue: Callable[[], bool] | None) -> bool:
|
||
return should_continue is None or bool(should_continue())
|
||
|
||
|
||
def _write_config(
|
||
exe: Path,
|
||
config: dict[str, Any],
|
||
runtime_dir: Path,
|
||
config_file: Path,
|
||
) -> dict[str, Any]:
|
||
"""Worker-only: persist the runtime JSON; return its redacted diagnostic copy."""
|
||
runtime_dir.mkdir(parents=True, exist_ok=True)
|
||
config_file.write_text(json.dumps(config, ensure_ascii=True, indent=2), encoding="utf-8")
|
||
return capture_runtime_config(exe, config)
|
||
|
||
|
||
#: Новый TUN не раньше, чем через столько секунд после остановки прежнего.
|
||
#: Windows создаёт Wintun-адаптер вслед за только что снятым медленно, а
|
||
#: «родной» sing-box, поднятый сразу, на win10 оставался без TCP-трафика
|
||
#: через туннель. Замер: без паузы — сбой почти всегда, 1 с — 1 из 4,
|
||
#: 2 с — 2 из 3 на переходе с VLESS с перезапуском winws2, 4 с — 0 из 5.
|
||
#: Прежде эту паузу случайно давало 10-секундное ожидание любого xftun.
|
||
TUN_RESTART_GAP_S = 4.0
|
||
|
||
|
||
def _kill_orphaned_blocking(exe: Path) -> bool:
|
||
"""Worker-only: kill orphaned sing-box processes that hold ports/TUN."""
|
||
try:
|
||
return bool(kill_processes_by_path(exe.name, exe, timeout=5, pump=False))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
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._tun_interface_name = ""
|
||
self._tun_stopped_at: float | None = None
|
||
self._defer_startup_failure = False
|
||
self._deferred_startup_failure: str | None = None
|
||
self._last_start_failure_retryable = False
|
||
self.diagnostic_config: dict[str, Any] | None = None
|
||
# Правила route последних запусков: диагностика видит и конфиг до правки.
|
||
self.diagnostic_route_history: deque[dict[str, Any]] = deque(maxlen=8)
|
||
|
||
@property
|
||
def is_running(self) -> bool:
|
||
return self._running
|
||
|
||
@property
|
||
def last_start_failure_retryable(self) -> bool:
|
||
return self._last_start_failure_retryable
|
||
|
||
@property
|
||
def process_alive(self) -> bool:
|
||
"""A spawned process exists, even if readiness was never confirmed."""
|
||
return self._process.state() != QProcess.ProcessState.NotRunning
|
||
|
||
def start(self, singbox_path: str, config: dict[str, Any]) -> bool:
|
||
# Синхронный драйвер тех же шагов — только для shutdown/тестов.
|
||
# Переходы идут через start_steps() в TransitionRunner.
|
||
return bool(run_steps_blocking(self.start_steps(singbox_path, config)))
|
||
|
||
def start_steps(
|
||
self,
|
||
singbox_path: str,
|
||
config: dict[str, Any],
|
||
*,
|
||
should_continue: Callable[[], bool] | None = None,
|
||
) -> TransitionSteps:
|
||
"""Start sing-box without blocking the GUI thread.
|
||
|
||
File writes, ``sing-box check``, the orphan scan and every readiness
|
||
probe run in the worker pool; waits are timer-driven steps.
|
||
``sing-box check`` runs only after a failed launch, once the process is
|
||
gone: a good config no longer waits for a separate core process, and a
|
||
bad one still reports the core's own explanation of the rejected field.
|
||
Never alongside a running core — for a TUN config the check builds the
|
||
same named Wintun interface and broke the live tunnel's TCP.
|
||
``should_continue`` is polled between steps: once it turns false the
|
||
half-started process is stopped and the start returns ``False``
|
||
without reporting a failure (a newer transition owns the outcome).
|
||
"""
|
||
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)
|
||
config_file = SINGBOX_CONFIG_FILE
|
||
runtime_dir = RUNTIME_DIR
|
||
|
||
self.diagnostic_config = yield run_in_worker(
|
||
lambda: _write_config(exe, config, runtime_dir, config_file)
|
||
)
|
||
entry = route_history_entry(self.diagnostic_config)
|
||
if entry is not None and (
|
||
not self.diagnostic_route_history
|
||
or self.diagnostic_route_history[-1]["route_rules"] != entry["route_rules"]
|
||
or self.diagnostic_route_history[-1]["route_final"] != entry["route_final"]
|
||
):
|
||
self.diagnostic_route_history.append(entry)
|
||
if not _still_wanted(should_continue):
|
||
return False
|
||
self._deferred_startup_failure = None
|
||
self._defer_startup_failure = True
|
||
try:
|
||
started = yield from self._launch_steps(
|
||
exe, config, config_file, tun_interface_name, should_continue
|
||
)
|
||
finally:
|
||
self._defer_startup_failure = False
|
||
deferred, self._deferred_startup_failure = self._deferred_startup_failure, None
|
||
if started or deferred is None:
|
||
return started
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
yield from self.stop_steps(expected=True)
|
||
config_ok, config_problem = yield run_in_worker(lambda: check_config(exe, config_file))
|
||
self.error.emit(deferred if config_ok else config_problem)
|
||
return False
|
||
|
||
def _launch_steps(
|
||
self,
|
||
exe: Path,
|
||
config: dict[str, Any],
|
||
config_file: Path,
|
||
tun_interface_name: str,
|
||
should_continue: Callable[[], bool] | None,
|
||
) -> TransitionSteps:
|
||
uses_tun = bool(tun_interface_name)
|
||
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
if not (yield from self.stop_steps(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.
|
||
killed = yield run_in_worker(lambda: _kill_orphaned_blocking(exe))
|
||
if killed:
|
||
yield sleep_ms(1000)
|
||
if not _still_wanted(should_continue):
|
||
return False
|
||
self._uses_tun = uses_tun
|
||
self._tun_interface_name = tun_interface_name or ""
|
||
|
||
# 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._spawn(exe, core_dir, config_file)
|
||
if not (yield wait_process_started(self._process, 4000)):
|
||
self._starting = False
|
||
self._report_startup_failure(f"failed to start sing-box process: {self._process.errorString()}")
|
||
return False
|
||
ready = yield from self._wait_until_proxy_ready_steps(config, should_continue=should_continue)
|
||
self._starting = False
|
||
if not ready:
|
||
return False
|
||
self._mark_running()
|
||
return True
|
||
|
||
if self._tun_stopped_at is not None:
|
||
remaining = TUN_RESTART_GAP_S - (time.monotonic() - self._tun_stopped_at)
|
||
if remaining > 0:
|
||
yield sleep_ms(int(remaining * 1000))
|
||
if not _still_wanted(should_continue):
|
||
self._starting = False
|
||
return False
|
||
|
||
# TUN mode retries while Windows releases a previous wintun adapter.
|
||
for attempt in range(3):
|
||
self._spawn(exe, core_dir, config_file)
|
||
if not (yield wait_process_started(self._process, 4000)):
|
||
self._starting = False
|
||
self._report_startup_failure(f"failed to start sing-box process: {self._process.errorString()}")
|
||
return False
|
||
|
||
ready = yield from self._wait_until_tun_ready_steps(
|
||
tun_interface_name, should_continue=should_continue
|
||
)
|
||
if ready:
|
||
self._starting = False
|
||
self._mark_running()
|
||
return True
|
||
if not _still_wanted(should_continue):
|
||
self._starting = False
|
||
yield from self.stop_steps(expected=True)
|
||
return False
|
||
|
||
exited = self._process.state() == QProcess.ProcessState.NotRunning
|
||
retryable = exited and self._startup_error_is_retryable()
|
||
if not exited:
|
||
yield from self.stop_steps(expected=True)
|
||
|
||
if retryable and attempt < 2:
|
||
yield from self._wait_tun_released_steps(tun_interface_name)
|
||
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
|
||
|
||
def _spawn(self, exe: Path, core_dir: Path, config_file: Path) -> None:
|
||
self._last_output_lines.clear()
|
||
self._process.setWorkingDirectory(str(core_dir))
|
||
self._process.setProgram(str(exe))
|
||
self._process.setArguments(["run", "-c", str(config_file), "-D", str(core_dir)])
|
||
self._process.start()
|
||
|
||
def stop(self, expected: bool = True) -> bool:
|
||
# Синхронный драйвер — shutdown и тесты; переходы используют stop_steps().
|
||
return bool(run_steps_blocking(self.stop_steps(expected=expected)))
|
||
|
||
def request_stop(self, expected: bool = True) -> None:
|
||
"""Close admission now without waiting (signal handlers).
|
||
|
||
``kill()`` returns immediately; ``finished`` arrives through the event
|
||
loop and updates the running state. A later ``start_steps`` waits for
|
||
that exit before reusing ports or the TUN adapter.
|
||
"""
|
||
if self._process.state() == QProcess.ProcessState.NotRunning:
|
||
return
|
||
self._stop_requested = expected
|
||
self._process.kill()
|
||
|
||
def stop_steps(self, expected: bool = True) -> TransitionSteps:
|
||
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
|
||
tun_name = self._tun_interface_name
|
||
self._stop_requested = expected
|
||
# На Windows консольный sing-box игнорирует terminate() (WM_CLOSE):
|
||
# прежние 500 мс «мягкой» остановки истекали всегда и только задерживали
|
||
# каждое отключение и переключение. Адаптер Wintun исчезает вместе с
|
||
# процессом, его освобождение ждёт _wait_tun_released_steps ниже.
|
||
self._process.kill()
|
||
yield wait_process_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:
|
||
yield from self._wait_tun_released_steps(tun_name)
|
||
self._tun_stopped_at = time.monotonic()
|
||
return True
|
||
|
||
@staticmethod
|
||
def _wait_tun_released_steps(interface_name: str = "", max_wait: float = 10.0) -> TransitionSteps:
|
||
"""Poll until this session's TUN adapter is released, up to max_wait seconds.
|
||
|
||
Ждём только свой адаптер: имя у каждого запуска новое (xftun + hex),
|
||
а «призраки» прежних сессий (Wintun убитого ядра остаётся в списке
|
||
выключенным) раньше держали здесь каждое отключение все 10 секунд.
|
||
"""
|
||
if os.name != "nt" or not interface_name:
|
||
return None
|
||
deadline = time.monotonic() + max(0.0, max_wait)
|
||
while time.monotonic() < deadline:
|
||
gone, fast_probe = yield run_in_worker(
|
||
lambda: SingBoxManager._probe_tun_adapter_gone(interface_name)
|
||
)
|
||
if gone:
|
||
return None
|
||
yield sleep_ms(LISTENER_POLL_MS if fast_probe else 300)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _probe_tun_adapter_gone(interface_name: str) -> tuple[bool, bool]:
|
||
"""Return (adapter is released, fast ctypes path was used). Worker-only."""
|
||
if win_netinfo.is_available():
|
||
try:
|
||
return (not win_netinfo.adapter_is_up(interface_name)), True
|
||
except Exception:
|
||
pass # fall back to netsh below
|
||
try:
|
||
result = run_text(
|
||
["netsh", "interface", "show", "interface", f"name={interface_name}"],
|
||
timeout=3,
|
||
creationflags=_CREATE_NO_WINDOW,
|
||
)
|
||
# Нет такого интерфейса (ненулевой код) или он уже не подключён.
|
||
text = result_output_text(result).lower()
|
||
if result.returncode != 0 or interface_name.lower() not in text:
|
||
return True, False
|
||
return any(word in text for word in ("disconnected", "отключ")), 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_steps(
|
||
self,
|
||
config: dict[str, Any],
|
||
max_wait: float = 15.0,
|
||
*,
|
||
should_continue: Callable[[], bool] | None = None,
|
||
) -> TransitionSteps:
|
||
port_roles = self._extract_proxy_port_roles(config)
|
||
port_credentials = self._extract_socks_credentials(config)
|
||
if not port_roles:
|
||
yield sleep_ms(750)
|
||
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
|
||
|
||
def probe_all() -> bool | None:
|
||
# None — порты ещё не открыты: подключаться рано, ответ дала таблица TCP.
|
||
if ports_listening(port_roles) is False:
|
||
return None
|
||
return all(
|
||
probe_listener_role(port, role, **(port_credentials.get(port) or {}))
|
||
for port, role in port_roles.items()
|
||
)
|
||
|
||
def pending_listeners() -> list[str]:
|
||
return [
|
||
f"{role} {port}"
|
||
for port, role in port_roles.items()
|
||
if not probe_listener_role(port, role, **(port_credentials.get(port) or {}))
|
||
]
|
||
|
||
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 not _still_wanted(should_continue):
|
||
self._starting = False
|
||
yield from self.stop_steps(expected=True)
|
||
return False
|
||
# Пробные connect'ы (до 0.35с на порт) — в worker-пуле.
|
||
ready = yield run_in_worker(probe_all)
|
||
if ready:
|
||
return True
|
||
yield sleep_ms(LISTENER_POLL_MS if ready is None else 100)
|
||
|
||
self._last_start_failure_retryable = True
|
||
pending = yield run_in_worker(pending_listeners)
|
||
# Диагностику ядра нужно снять до stop(): штатная остановка стирает
|
||
# контекст последних строк как «ожидаемый» выход.
|
||
last_output = next(
|
||
(line for line in reversed(self._last_output_lines) if line.strip()), ""
|
||
)
|
||
yield from self.stop_steps(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_steps(
|
||
self,
|
||
tun_interface_name: str,
|
||
max_wait: float = 18.0,
|
||
*,
|
||
should_continue: Callable[[], bool] | None = None,
|
||
) -> TransitionSteps:
|
||
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
|
||
if not _still_wanted(should_continue):
|
||
return False
|
||
has_ipv4, fast_probe = yield run_in_worker(
|
||
lambda: self._probe_tun_interface_has_ipv4(tun_interface_name)
|
||
)
|
||
if has_ipv4:
|
||
return True
|
||
yield sleep_ms(100 if fast_probe else 250)
|
||
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. Worker-only."""
|
||
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(
|
||
["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
|
||
if getattr(self, "_defer_startup_failure", False):
|
||
# Ядро отвергает негодную конфигурацию уже после старта процесса,
|
||
# печатая причину в свой лог вперемешку с ANSI-кодами. Сообщение
|
||
# отдаст start_steps, сверив его с ответом `sing-box check`: там
|
||
# объяснение (какое поле не принято) понятнее кода выхода.
|
||
self._deferred_startup_failure = message
|
||
return
|
||
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()
|