zapret-kvn/xray_fluent/engines/singbox/manager.py
loop-uh d5a4715771
Some checks failed
Windows project source guards / test (push) Has been cancelled
feat: use official Amnezia transport and organize runtime modules
2026-09-06 00:14:08 +03:00

550 lines
21 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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()