loop-uh
18f49a1573
Some checks failed
Windows project source guards / test (push) Failing after 26s
После замера скорости через тот же временный прокси к YouTube, Discord, Telegram и ещё восьми сервисам идёт серия запросов по одному соединению. По времени до первого байта решается: быстро, медленно, с паузами, открывается не каждый адрес или не открывается. Итог хранится в узле и показан строкой «Через этот узел» в карточке сервера со словами о проблемах. Раньше узел проверялся одним байтом generate_204 и скоростью файла. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
4282 lines
197 KiB
Python
4282 lines
197 KiB
Python
from __future__ import annotations
|
||
|
||
from copy import deepcopy
|
||
from dataclasses import dataclass
|
||
import hashlib
|
||
import logging
|
||
import random
|
||
import re
|
||
import socket
|
||
import time
|
||
from datetime import datetime, timezone
|
||
import json
|
||
from logging.handlers import RotatingFileHandler
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Any, Callable
|
||
|
||
from PyQt6.QtCore import QObject, QTimer, pyqtSignal
|
||
|
||
from .config import (
|
||
SingboxDocumentCache,
|
||
apply_singbox_config_text as apply_singbox_config_text_operation,
|
||
apply_xray_config_text as apply_xray_config_text_operation,
|
||
apply_xray_tun_loop_prevention as apply_xray_tun_loop_prevention_operation,
|
||
build_runtime_xray_config as build_runtime_xray_config_operation,
|
||
collect_xray_inbound_ports,
|
||
config_has_proxy_outbound,
|
||
default_singbox_config_text,
|
||
default_xray_config_text,
|
||
ensure_active_config as ensure_active_config_operation,
|
||
ensure_dict,
|
||
ensure_list,
|
||
ensure_xray_metrics_contract as ensure_xray_metrics_contract_operation,
|
||
ensure_xray_tun_contract as ensure_xray_tun_contract_operation,
|
||
extract_xray_runtime_ports,
|
||
format_json_error_message,
|
||
get_active_config_name as get_active_config_name_operation,
|
||
get_active_config_path as get_active_config_path_operation,
|
||
get_active_template_path as get_active_template_path_operation,
|
||
import_template as import_template_operation,
|
||
infer_singbox_outbound_endpoint,
|
||
infer_singbox_ping_target,
|
||
infer_xray_outbound_endpoint,
|
||
infer_xray_ping_target,
|
||
inspect_active_xray_config as inspect_active_xray_config_operation,
|
||
is_local_runtime_host,
|
||
load_active_config_text as load_active_config_text_operation,
|
||
load_config_text as load_config_text_operation,
|
||
normalize_relative_json_path,
|
||
replace_or_append_tagged,
|
||
reset_active_config_to_template as reset_active_config_to_template_operation,
|
||
resolve_profile_path,
|
||
save_config_text as save_config_text_operation,
|
||
validate_json_text,
|
||
xray_outbound_is_loop_protected as xray_outbound_is_loop_protected_operation,
|
||
)
|
||
from .nodes import (
|
||
bulk_update_nodes as bulk_update_nodes_operation,
|
||
check_auto_switch as check_auto_switch_operation,
|
||
detect_countries_sync as detect_countries_sync_operation,
|
||
get_all_groups as get_all_groups_operation,
|
||
get_all_tags as get_all_tags_operation,
|
||
get_fastest_alive_node as get_fastest_alive_node_operation,
|
||
get_next_node_for_auto_switch as get_next_node_for_auto_switch_operation,
|
||
get_node_by_id as get_node_by_id_operation,
|
||
import_nodes_from_text as import_nodes_from_text_operation,
|
||
log_selection_change,
|
||
selection_snapshot,
|
||
on_countries_resolved as on_countries_resolved_operation,
|
||
prepare_node_for_runtime as prepare_node_for_runtime_operation,
|
||
remove_nodes as remove_nodes_operation,
|
||
reorder_nodes as reorder_nodes_operation,
|
||
set_selected_node as set_selected_node_operation,
|
||
start_country_ip_resolution as start_country_ip_resolution_operation,
|
||
update_node as update_node_operation,
|
||
)
|
||
from .selection_source import SelectionSource
|
||
from .auto_switch_service import (
|
||
begin_auto_switch_warmup,
|
||
transport_kind_for_node,
|
||
)
|
||
from .smart_switch_service import (
|
||
SmartSwitchState,
|
||
cancel_smart_check,
|
||
check_smart_switch as check_smart_switch_operation,
|
||
on_candidate_result as on_smart_candidate_result_operation,
|
||
on_candidates_done as on_smart_candidates_done_operation,
|
||
on_current_probe_measured as on_smart_probe_measured_operation,
|
||
)
|
||
from ..diagnostics.runtime_errors import (
|
||
RepeatedNoticeFilter,
|
||
RuntimeErrorJournal,
|
||
core_failure,
|
||
is_core_error_line,
|
||
is_deprecation_notice,
|
||
)
|
||
from ..engines.hysteria.runtime_contract import (
|
||
AUTOMATIC_SWITCH_FAILURES,
|
||
SECURITY_FAILURES,
|
||
HysteriaFailureCode,
|
||
HysteriaRuntimeState,
|
||
HysteriaTransitionContract,
|
||
classify_hysteria_uri,
|
||
node_is_maintenance,
|
||
)
|
||
from .singbox_config_recovery import (
|
||
SingboxConfigRepair,
|
||
repair_singbox_config_file,
|
||
try_repair_singbox_config_text,
|
||
)
|
||
from .runtime import (
|
||
ActiveSessionSnapshot,
|
||
TransitionContext,
|
||
XrayRuntimeConfig,
|
||
build_active_session_snapshot,
|
||
can_apply_proxy_runtime_change as can_apply_proxy_runtime_change_rule,
|
||
can_proxy_hot_swap as can_proxy_hot_swap_rule,
|
||
can_tun_hot_swap as can_tun_hot_swap_rule,
|
||
cancel_ping as cancel_ping_operation,
|
||
cancel_speed_test as cancel_speed_test_operation,
|
||
cleanup_connection_runtime_state as cleanup_connection_runtime_state_operation,
|
||
compute_transition_action,
|
||
connect_selected as connect_selected_operation,
|
||
connect_selected_steps as connect_selected_steps_operation,
|
||
disconnect_current as disconnect_current_operation,
|
||
disconnect_current_steps as disconnect_current_steps_operation,
|
||
handle_unexpected_disconnect as handle_unexpected_disconnect_operation,
|
||
handle_unexpected_disconnect_steps as handle_unexpected_disconnect_steps_operation,
|
||
needs_transition,
|
||
on_connectivity_result as on_connectivity_result_operation,
|
||
on_core_state_changed as on_core_state_changed_operation,
|
||
on_live_metrics as on_live_metrics_operation,
|
||
on_ping_measured as on_ping_measured_operation,
|
||
on_ping_progress as on_ping_progress_operation,
|
||
on_services_result as on_services_result_operation,
|
||
on_speed_complete as on_speed_complete_operation,
|
||
on_speed_node_progress as on_speed_node_progress_operation,
|
||
on_speed_progress as on_speed_progress_operation,
|
||
on_speed_result as on_speed_result_operation,
|
||
on_xray_update_worker_done as on_xray_update_worker_done_operation,
|
||
ping_nodes as ping_nodes_operation,
|
||
reconnect as reconnect_operation,
|
||
reconnect_steps as reconnect_steps_operation,
|
||
routing_signature as routing_signature_operation,
|
||
run_xray_core_update as run_xray_core_update_operation,
|
||
shutdown as shutdown_operation,
|
||
signature as signature_operation,
|
||
speed_test_nodes as speed_test_nodes_operation,
|
||
start_metrics_worker as start_metrics_worker_operation,
|
||
stop_active_connection_processes as stop_active_connection_processes_operation,
|
||
stop_active_connection_processes_steps as stop_active_connection_processes_steps_operation,
|
||
stop_metrics_worker as stop_metrics_worker_operation,
|
||
system_proxy_bypass_lan as system_proxy_bypass_lan_operation,
|
||
test_connectivity as test_connectivity_operation,
|
||
transition_request_delay_ms,
|
||
transition_signature as transition_signature_operation,
|
||
transition_status_text,
|
||
tun_layer_signature as tun_layer_signature_operation,
|
||
xray_layer_signature as xray_layer_signature_operation,
|
||
)
|
||
from .signature_service import SingboxPlanFacts
|
||
from .subscription_service import (
|
||
apply_not_modified,
|
||
hide_subscription_node as hide_subscription_node_operation,
|
||
mark_subscription_failure,
|
||
reconcile_subscription,
|
||
remove_subscription as remove_subscription_operation,
|
||
subscription_due,
|
||
)
|
||
from .async_steps import TransitionRunner, TransitionSteps, run_in_worker, run_steps_blocking, sleep_ms
|
||
from .rotation_service import (
|
||
RotationPlan,
|
||
build_rotation_plan,
|
||
pick_next_node,
|
||
rotation_interval_ms,
|
||
)
|
||
from .outbound_pool_service import (
|
||
SINGBOX_SELECTOR_TAG,
|
||
XRAY_BALANCER_TAG,
|
||
XrayOutboundPool,
|
||
build_xray_outbound_pool,
|
||
)
|
||
from ..profiles.country_flags import CountryResolver
|
||
from ..network.background_workers import (
|
||
StateWriter,
|
||
SubscriptionUpdateWorker,
|
||
)
|
||
from ..engines.xray import (
|
||
XrayManager,
|
||
XrayTunRouteManager,
|
||
apply_balancer_override,
|
||
build_xray_config,
|
||
get_windows_default_route_context,
|
||
get_xray_version,
|
||
)
|
||
from ..engines.hysteria import HysteriaManager
|
||
from ..engines.amnezia.manager import AmneziaManager
|
||
from ..engines.singbox import (
|
||
SingBoxManager,
|
||
classify_node_for_singbox,
|
||
get_singbox_version,
|
||
parse_singbox_document,
|
||
plan_singbox_proxy_runtime,
|
||
plan_singbox_runtime,
|
||
restart_proxy_runtime as restart_singbox_proxy_runtime_operation,
|
||
restart_proxy_runtime_steps as restart_singbox_proxy_runtime_steps_operation,
|
||
restart_runtime as restart_singbox_runtime_operation,
|
||
restart_runtime_steps as restart_singbox_runtime_steps_operation,
|
||
SingboxDocumentState,
|
||
SingboxRuntimePlan,
|
||
select_outbound as select_singbox_outbound,
|
||
select_outbound_when_ready_steps as select_singbox_outbound_when_ready_steps,
|
||
)
|
||
from ..constants import (
|
||
APP_NAME,
|
||
DEFAULT_HTTP_PORT,
|
||
DEFAULT_SOCKS_PORT,
|
||
DEFAULT_XRAY_STATS_API_PORT,
|
||
LOG_DIR,
|
||
PROXY_HOST,
|
||
ROUTING_MODES,
|
||
SINGBOX_CLASH_API_PORT,
|
||
SINGBOX_PROVIDER_FILE,
|
||
SINGBOX_CONFIGS_DIR,
|
||
SINGBOX_DEFAULT_CONFIG_NAME,
|
||
SINGBOX_TEMPLATES_DIR,
|
||
XRAY_CONFIGS_DIR,
|
||
XRAY_DEFAULT_CONFIG_NAME,
|
||
XRAY_TUN_DEFAULT_INTERFACE_NAME,
|
||
XRAY_TEMPLATES_DIR,
|
||
STATE_SCHEMA_VERSION,
|
||
)
|
||
from ..diagnostics.export import collect_runtime_diagnostics, export_diagnostics
|
||
from ..profiles.models import (
|
||
AppSettings,
|
||
AppState,
|
||
Node,
|
||
RoutingSettings,
|
||
Subscription,
|
||
SubscriptionUpdateResult,
|
||
ZapretTargetSettings,
|
||
clamp_subscriptions_check_interval,
|
||
normalize_subscription_warnings,
|
||
utc_now_iso,
|
||
)
|
||
from ..importer.subscription_http import (
|
||
normalize_client_profile,
|
||
resolve_subscription_source,
|
||
validate_hwid,
|
||
)
|
||
from ..importer.subscription_parser import validate_filter_patterns
|
||
from ..network.bootstrap_dns import refresh_async as refresh_bootstrap_address
|
||
from ..network.network_monitor import NetworkMonitor
|
||
from ..network.ping_service import PingService
|
||
from ..platform.windows.proxy_manager import PROXY_EXECUTOR, ProxyManager, SystemProxyState
|
||
from ..platform.windows.security import create_password_hash, get_idle_seconds, verify_password
|
||
from ..diagnostics.runtime_logging import (
|
||
ServerAddressLogFilter,
|
||
redact_server_addresses,
|
||
register_server_nodes,
|
||
RuntimeLogContext,
|
||
RuntimeNodeIdentity,
|
||
contextualize_runtime_log,
|
||
identities_for_tags,
|
||
runtime_mapping_lines,
|
||
)
|
||
from ..profiles.storage import PassphraseRequired, StateStorage
|
||
from ..platform.windows import startup as windows_startup
|
||
from ..platform.windows.subprocess_utils import result_output_text, run_text
|
||
from ..diagnostics.traffic_history import TrafficHistoryFileSink, TrafficHistoryStorage
|
||
from .server_bypass import ServerBypass
|
||
from ..engines.zapret.manager import ZapretManager
|
||
|
||
if TYPE_CHECKING:
|
||
from ..profiles.country_flags import CountryResolver as CountryResolverType
|
||
from ..network.connectivity_test import ConnectivityTestWorker
|
||
from ..engines.xray import XrayCoreUpdateResult, XrayCoreUpdateWorker
|
||
from ..diagnostics.live_metrics_worker import LiveMetricsWorker
|
||
from ..network.speed_test_worker import SpeedTestWorker
|
||
|
||
|
||
def _increment_int(value: object) -> int:
|
||
"""Advance a generation counter, including lightweight Mock controllers in tests."""
|
||
|
||
return value + 1 if isinstance(value, int) else 1
|
||
|
||
|
||
def _find_free_api_port(preferred: int | None = None, excluded: set[int] | None = None) -> int:
|
||
"""Find a free TCP port near *preferred* for the xray stats API."""
|
||
if preferred is None:
|
||
preferred = DEFAULT_XRAY_STATS_API_PORT
|
||
for port in range(preferred, preferred + 100):
|
||
if excluded and port in excluded:
|
||
continue
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||
try:
|
||
s.bind(("127.0.0.1", port))
|
||
return port
|
||
except OSError:
|
||
continue
|
||
raise RuntimeError(f"No free port in range {preferred}-{preferred + 100}")
|
||
|
||
|
||
def _process_alive(manager: object) -> bool:
|
||
"""Spawned-but-not-ready processes are invisible to ``is_running``."""
|
||
return getattr(manager, "process_alive", False) is True
|
||
|
||
|
||
def _log_transition_profile(controller: Any, kind: str, runner: TransitionRunner) -> None:
|
||
"""Одна строка [perf] на переход: где ушло время подключения/переключения."""
|
||
log = getattr(controller, "_log", None)
|
||
profile = getattr(runner, "profile", None)
|
||
if log is None or profile is None:
|
||
return
|
||
outcome = "cancelled" if runner.cancelled else ("error" if runner.error is not None else (
|
||
"ok" if runner.result else "failed"
|
||
))
|
||
requested = getattr(runner, "requested_at", None)
|
||
waited = ""
|
||
if requested is not None:
|
||
waited = f" since_request={(time.perf_counter() - requested) * 1000.0:.0f}ms"
|
||
log(f"[perf] {kind} {outcome}{waited} {profile.summary()}")
|
||
|
||
|
||
def _write_json_atomically(path: Path, payload: object) -> None:
|
||
"""Worker-only: serialize and atomically replace a runtime JSON file."""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
temporary = path.with_suffix(".json.tmp")
|
||
temporary.write_text(json.dumps(payload, ensure_ascii=True, indent=2), encoding="utf-8")
|
||
temporary.replace(path)
|
||
|
||
|
||
# AC6 (startup-subscription-settings): hard fallback for the
|
||
# "after_subscriptions" startup order — auto-connect proceeds after this delay
|
||
# even if startup subscription updates hang or fail (spec limit: <= 30 s).
|
||
STARTUP_CONNECT_FALLBACK_MS = 20_000
|
||
|
||
_XRAY_METRICS_API_TAG = "__app_metrics_api"
|
||
_XRAY_METRICS_API_INBOUND_TAG = "__app_metrics_api_in"
|
||
_XRAY_TUN_INBOUND_TAG = "__app_tun_in"
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class HotSwitchPlan:
|
||
"""Immutable outcome of the pure hot-switch feasibility checks (П2).
|
||
|
||
Captured on the GUI thread in the same tick the switch attempt starts, so
|
||
the async steps operate on a consistent snapshot of session/pool state.
|
||
"""
|
||
|
||
node: Node
|
||
session: ActiveSessionSnapshot
|
||
tags: dict[str, str]
|
||
tag: str
|
||
control_core: str
|
||
previous_tag: str
|
||
|
||
|
||
class AppController(QObject):
|
||
runtime_errors_changed = pyqtSignal(object)
|
||
# Снимок реестра после каждой операции системного прокси (очередь WinAPI
|
||
# асинхронная — без этого UI держал старое «Вкл» до следующего события).
|
||
system_proxy_state_changed = pyqtSignal(object)
|
||
_proxy_operation_done = pyqtSignal()
|
||
nodes_changed = pyqtSignal(object)
|
||
countries_changed = pyqtSignal(object)
|
||
subscriptions_changed = pyqtSignal(object)
|
||
subscription_update_started = pyqtSignal(str)
|
||
subscription_update_progress = pyqtSignal(str, str)
|
||
subscription_update_finished = pyqtSignal(object)
|
||
selection_changed = pyqtSignal(object)
|
||
connection_changed = pyqtSignal(bool)
|
||
connection_status_changed = pyqtSignal(str, str)
|
||
routing_changed = pyqtSignal(object)
|
||
settings_changed = pyqtSignal(object)
|
||
log_line = pyqtSignal(str)
|
||
status = pyqtSignal(str, str)
|
||
bulk_task_progress = pyqtSignal(str, int, int, bool) # task, current, total, completed
|
||
ping_updated = pyqtSignal(str, object)
|
||
speed_updated = pyqtSignal(str, object, bool) # node_id, speed_mbps, is_alive
|
||
speed_progress_updated = pyqtSignal(str, int) # node_id, percent
|
||
speed_test_cancelled = pyqtSignal(int, int) # completed, total
|
||
connectivity_test_done = pyqtSignal(bool, str, object)
|
||
live_metrics_updated = pyqtSignal(object)
|
||
xray_update_result = pyqtSignal(object)
|
||
lock_state_changed = pyqtSignal(bool)
|
||
passphrase_required = pyqtSignal()
|
||
auto_switch_triggered = pyqtSignal(str) # node name we're switching to
|
||
transition_state_changed = pyqtSignal(bool, str)
|
||
# Строки лога из рабочих потоков (очередь прокси) — в GUI-поток.
|
||
_background_log = pyqtSignal(str)
|
||
|
||
def __init__(self, parent: QObject | None = None):
|
||
super().__init__(parent)
|
||
self.storage = StateStorage()
|
||
self.xray = XrayManager(self)
|
||
self.singbox = SingBoxManager(self)
|
||
self.hysteria = HysteriaManager(self)
|
||
self.amnezia = self._new_amnezia_manager()
|
||
self._amnezia_target_generation = 0
|
||
self._xray_tun_routes = XrayTunRouteManager(self)
|
||
self.zapret = ZapretManager(self)
|
||
self.bypass = ServerBypass(self)
|
||
self.proxy = ProxyManager()
|
||
self.network_monitor = NetworkMonitor(parent=self)
|
||
|
||
self.state = AppState()
|
||
self._node_lookup_source_id = 0
|
||
self._node_lookup_size = -1
|
||
self._node_by_id: dict[str, Node] = {}
|
||
self.recent_logs: list[str] = []
|
||
self.runtime_errors = RuntimeErrorJournal()
|
||
self.connected = False
|
||
self.locked = False
|
||
|
||
# --- File logger (5 MB × 3 rotated files in data/logs/) ---
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
self._logger = logging.getLogger("xray_fluent")
|
||
self._logger.setLevel(logging.DEBUG)
|
||
if not self._logger.handlers:
|
||
handler = RotatingFileHandler(
|
||
LOG_DIR / "app.log",
|
||
maxBytes=5 * 1024 * 1024,
|
||
backupCount=3,
|
||
encoding="utf-8",
|
||
)
|
||
handler.setFormatter(logging.Formatter("%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
|
||
handler.addFilter(ServerAddressLogFilter())
|
||
self._logger.addHandler(handler)
|
||
# Адреса всех узлов (загрузка, подписки, правка) маскируются в логах
|
||
# и экспорте как «<сервер node_ref>».
|
||
self.nodes_changed.connect(register_server_nodes)
|
||
|
||
self._country_resolver: CountryResolver | None = None
|
||
# Пинг — долгоживущая служба без QThread; короткие QThread-воркеры
|
||
# живут в _background_workers до своего finished (worker_keeper).
|
||
self.ping = PingService(self)
|
||
self.ping.measured.connect(self._on_ping_measured)
|
||
self.ping.progress.connect(self._on_ping_progress)
|
||
self._background_workers: list = []
|
||
self._speed_worker: SpeedTestWorker | None = None
|
||
self._connectivity_worker: ConnectivityTestWorker | None = None
|
||
self._metrics_worker: LiveMetricsWorker | None = None
|
||
self._xray_update_worker: XrayCoreUpdateWorker | None = None
|
||
self._subscription_workers: dict[str, SubscriptionUpdateWorker] = {}
|
||
self._subscription_update_queue: list[
|
||
tuple[Subscription, str, bool, bool, bool]
|
||
] = []
|
||
self._subscription_queued_ids: set[str] = set()
|
||
self._pending_subscription_additions: dict[str, Subscription] = {}
|
||
self._subscription_check_ids: set[str] = set()
|
||
self._singbox_documents = SingboxDocumentCache()
|
||
self._speed_total = 0
|
||
self._speed_completed = 0
|
||
self._xray_update_silent = False
|
||
self._reconnect_after_xray_update = False
|
||
self._reconnecting = False
|
||
self._connecting = False
|
||
self._disconnecting = False
|
||
self._cleaning_connection_state = False
|
||
self._switching = False # suppress intermediate UI updates during stop→start
|
||
self._active_core: str = "singbox" # "xray" | "singbox" | "tun2socks"
|
||
self._protect_ss_port: int = 0
|
||
self._protect_ss_password: str = ""
|
||
self._tun2socks_proxy_username: str = ""
|
||
self._tun2socks_proxy_password: str = ""
|
||
self._xray_api_port: int = 0
|
||
self._singbox_clash_api_port: int = 0
|
||
self._traffic_history = TrafficHistoryStorage(load=False)
|
||
self._traffic_save_counter = 0
|
||
|
||
# --- Auto-switch state ---
|
||
self._auto_switch_last_switch: float = 0.0 # monotonic timestamp of last auto-switch
|
||
self._auto_switch_cycle_attempts: int = 0
|
||
self._auto_switch_exhausted: bool = False
|
||
self._auto_switch_transitioning: bool = False
|
||
self._auto_switch_link_down_since: float = 0.0 # monotonic ts of first failed active-node ping
|
||
self._auto_switch_manual_hold: bool = False
|
||
self._auto_switch_warmup_until: float = 0.0
|
||
self._auto_switch_health_node_id: str | None = None
|
||
# «Умная проверка» низкой скорости (application/smart_switch_service.py).
|
||
self._smart_switch = SmartSwitchState()
|
||
self._smart_switch_manual_at: float = 0.0 # monotonic ts ручного выбора сервера
|
||
self._active_session: ActiveSessionSnapshot | None = None
|
||
self._desired_connected = False
|
||
self._transition_active = False
|
||
self._transition_scheduled = False
|
||
self._transition_pending = False
|
||
self._transition_reason = ""
|
||
self._transition_generation = 0
|
||
self._blocked_transition_signature = ""
|
||
self._transition_runner: TransitionRunner | None = None
|
||
# Подготовка выбранного сервера (DNS/Zapret) выполняется один раз на
|
||
# поколение запроса; клик во время перехода откладывает её до конца.
|
||
self._protection_prepared_generation = 0
|
||
# Процессы прерванного перехода гасятся отдельным шагом очереди.
|
||
self._cleanup_pending = False
|
||
# Поля плана для сигнатур (без портов): ключ — документ, режим, нода, пул.
|
||
self._plan_facts_cache: dict[tuple, SingboxPlanFacts | None] = {}
|
||
# Следующий disconnect из очереди — без статуса «Отключено»
|
||
# (ошибку, объяснившую остановку, уже показали).
|
||
self._quiet_disconnect = False
|
||
# П2 (AC5/AC8): асинхронный горячий свитч — control-plane I/O в воркере,
|
||
# generation-сериализация запросов, устаревший результат отбрасывается.
|
||
self._hot_switch_runner: TransitionRunner | None = None
|
||
self._hot_switch_generation = 0
|
||
self._hot_switch_pending = False
|
||
# П3 (AC9/AC10): кэш пула outbound'ов по идентичности списка нод.
|
||
self._xray_outbound_pool_cache: XrayOutboundPool | None = None
|
||
self._xray_outbound_pool_cache_key: tuple | None = None
|
||
# Серверы, которые запущенное ядро Xray отвергло при разборе конфига
|
||
# (node_id -> (id(node.outbound), причина)). Пока outbound не заменён,
|
||
# узел не возвращается в пул: один такой outbound не должен снова
|
||
# останавливать запуск для всех.
|
||
self._xray_pool_quarantine: dict[str, tuple[int, str]] = {}
|
||
self._xray_pool_quarantine_version = 0
|
||
self._core_log_contexts: dict[str, RuntimeLogContext] = {}
|
||
self._core_notices = RepeatedNoticeFilter()
|
||
self._active_singbox_plan: SingboxRuntimePlan | None = None
|
||
self._session_generation = 0
|
||
self._front_process_generation = 0
|
||
self._front_target_generation = 0
|
||
self._relay_credentials_generation = 0
|
||
self._hysteria_process_generation = 0
|
||
self._hysteria_active_generation = 0
|
||
self._hysteria_failure_episode_id = 0
|
||
self._hysteria_last_failure_code: HysteriaFailureCode | None = None
|
||
self._hysteria_automatic_switch_attempted = False
|
||
self._hysteria_recovery_active = False
|
||
self._pending_transport_node_id: str | None = None
|
||
self._hysteria_cooldown_until: dict[str, float] = {}
|
||
self._hysteria_failure_started_at = 0.0
|
||
self._hysteria_return_episode = 0
|
||
self._hysteria_contract = HysteriaTransitionContract()
|
||
|
||
self.xray.log_received.connect(self._on_xray_log)
|
||
self.xray.error.connect(self._on_xray_error)
|
||
self.xray.state_changed.connect(self._on_core_state_changed)
|
||
self.xray.stopped.connect(lambda code: self._on_core_stopped("xray", code))
|
||
|
||
self.singbox.log_received.connect(lambda line: self._on_core_log("sing-box", line))
|
||
self.singbox.error.connect(lambda message: self._on_core_error("sing-box", message))
|
||
self.singbox.state_changed.connect(self._on_core_state_changed)
|
||
self.singbox.stopped.connect(lambda code: self._on_core_stopped("singbox", code))
|
||
|
||
self._bind_hysteria_manager(self.hysteria)
|
||
|
||
self._xray_tun_routes.log_received.connect(self._on_xray_log)
|
||
|
||
self.network_monitor.network_changed.connect(self._on_network_changed)
|
||
self._background_log.connect(self._log)
|
||
self._proxy_operation_done.connect(self._emit_system_proxy_state)
|
||
self.connection_changed.connect(lambda _: self._sync_rotation_timer())
|
||
# П3 (AC10): любое изменение нод инвалидирует кэш пула явно — идентичность
|
||
# списка не видит in-place правок содержимого outbound-словаря.
|
||
self.nodes_changed.connect(lambda _nodes: self._invalidate_xray_outbound_pool_cache())
|
||
self.nodes_changed.connect(lambda _nodes: self._invalidate_plan_facts_cache())
|
||
|
||
self._lock_timer = QTimer(self)
|
||
self._lock_timer.setInterval(15_000)
|
||
self._lock_timer.timeout.connect(self._check_auto_lock)
|
||
self._save_timer = QTimer(self)
|
||
self._save_timer.setSingleShot(True)
|
||
self._save_timer.setInterval(250)
|
||
self._save_timer.timeout.connect(self._flush_scheduled_save)
|
||
self._save_pending = False
|
||
self._state_writer = StateWriter(self.storage, parent=self)
|
||
self._state_writer.failed.connect(lambda message: self._log(f"[storage] background save failed: {message}"))
|
||
self._traffic_writer = StateWriter(TrafficHistoryFileSink(), parent=self)
|
||
self._traffic_writer.failed.connect(lambda message: self._log(f"[traffic] background save failed: {message}"))
|
||
self._traffic_history.set_writer(self._traffic_writer)
|
||
self._transition_timer = QTimer(self)
|
||
self._transition_timer.setSingleShot(True)
|
||
self._transition_timer.timeout.connect(self._drain_transition_queue)
|
||
self._subscription_timer = QTimer(self)
|
||
self._subscription_timer.setInterval(15 * 60 * 1000)
|
||
self._subscription_timer.timeout.connect(self._check_due_subscriptions)
|
||
# AC6: deferred auto-connect for startup_connect_order="after_subscriptions"
|
||
self._startup_connect_pending = False
|
||
self._startup_connect_timer = QTimer(self)
|
||
self._startup_connect_timer.setSingleShot(True)
|
||
self._startup_connect_timer.setInterval(STARTUP_CONNECT_FALLBACK_MS)
|
||
self._startup_connect_timer.timeout.connect(self._finish_startup_connect)
|
||
self._rotation_timer = QTimer(self)
|
||
self._rotation_timer.setSingleShot(True)
|
||
self._rotation_timer.timeout.connect(self._on_rotation_tick)
|
||
self._rotation_rng = random.Random()
|
||
self._rotation_pool_logged = ""
|
||
self._rotation_running = False
|
||
|
||
def load(self, state=None, history=None) -> bool:
|
||
try:
|
||
self.state = state if state is not None else self.storage.load()
|
||
self._traffic_history = history if history is not None else TrafficHistoryStorage()
|
||
self._traffic_history.set_writer(self._traffic_writer)
|
||
except PassphraseRequired:
|
||
self.passphrase_required.emit()
|
||
return False
|
||
|
||
self._detect_countries_sync()
|
||
self._migrate_sort_order()
|
||
if self.state.schema_version != STATE_SCHEMA_VERSION:
|
||
self.state.schema_version = STATE_SCHEMA_VERSION
|
||
self.schedule_save()
|
||
self.nodes_changed.emit(self.state.nodes)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.selection_changed.emit(self.selected_node)
|
||
self.routing_changed.emit(self.state.routing)
|
||
self.settings_changed.emit(self.state.settings)
|
||
QTimer.singleShot(500, self._start_country_ip_resolution)
|
||
|
||
self.network_monitor.start()
|
||
self._lock_timer.start()
|
||
self._apply_subscription_timer_interval()
|
||
self._subscription_timer.start()
|
||
if self._startup_subscription_check_enabled():
|
||
QTimer.singleShot(30_000, self._check_due_subscriptions)
|
||
return True
|
||
|
||
def set_data_passphrase(self, passphrase: str) -> None:
|
||
self.storage.passphrase = passphrase
|
||
self.save()
|
||
self.status.emit("success", "Шифрование данных включено")
|
||
|
||
def clear_data_passphrase(self) -> None:
|
||
self.storage.passphrase = ""
|
||
self.save()
|
||
self.status.emit("info", "Шифрование данных отключено (портативный режим)")
|
||
|
||
def is_data_encrypted(self) -> bool:
|
||
return self.storage.is_encrypted()
|
||
|
||
def save(self) -> None:
|
||
"""Сохранить состояние без блокировки GUI-потока.
|
||
|
||
Здесь только снимок в текст (единицы мс даже на тысячах серверов);
|
||
шифрование и запись делает ``StateWriter`` в своём потоке.
|
||
"""
|
||
if self._save_timer.isActive():
|
||
self._save_timer.stop()
|
||
self._save_pending = False
|
||
passphrase = self.storage.passphrase
|
||
self._state_writer.submit(self.storage.serialize_state(self.state), passphrase)
|
||
self.storage._known_encrypted = bool(passphrase)
|
||
|
||
def save_blocking(self) -> None:
|
||
"""Сохранить и дождаться записи — только при выходе из приложения."""
|
||
self.save()
|
||
if not self._state_writer.close():
|
||
self._log("[storage] state write did not finish before exit")
|
||
if not self._traffic_writer.close():
|
||
self._log("[traffic] history write did not finish before exit")
|
||
|
||
def schedule_save(self) -> None:
|
||
self._save_pending = True
|
||
self._save_timer.start()
|
||
|
||
def _flush_scheduled_save(self) -> None:
|
||
if self._save_pending:
|
||
self.save()
|
||
|
||
@staticmethod
|
||
def _signature(payload: object) -> str:
|
||
return signature_operation(payload)
|
||
|
||
def _routing_signature(self, routing: RoutingSettings | None = None) -> str:
|
||
return routing_signature_operation(self, routing)
|
||
|
||
def is_singbox_tun_mode(self, settings: AppSettings | None = None) -> bool:
|
||
settings = settings or self.state.settings
|
||
return bool(settings.tun_mode)
|
||
|
||
def is_singbox_proxy_mode(self, settings: AppSettings | None = None) -> bool:
|
||
settings = settings or self.state.settings
|
||
return not settings.tun_mode
|
||
|
||
def is_singbox_editor_mode(self, settings: AppSettings | None = None) -> bool:
|
||
return True
|
||
|
||
def is_xray_tun_mode(self, settings: AppSettings | None = None) -> bool:
|
||
return False
|
||
|
||
def is_tun2socks_mode(self, settings: AppSettings | None = None) -> bool:
|
||
return False
|
||
|
||
def uses_xray_raw_config(self, settings: AppSettings | None = None) -> bool:
|
||
return False
|
||
|
||
def _can_connect_without_selected_node(self, settings: AppSettings | None = None) -> bool:
|
||
_, _, has_proxy_outbound = self._inspect_active_singbox_config()
|
||
return not has_proxy_outbound
|
||
|
||
def _system_proxy_bypass_lan(self, settings: AppSettings | None = None) -> bool:
|
||
return system_proxy_bypass_lan_operation(self, settings)
|
||
|
||
def get_singbox_config_dir(self) -> Path:
|
||
SINGBOX_CONFIGS_DIR.mkdir(parents=True, exist_ok=True)
|
||
return SINGBOX_CONFIGS_DIR
|
||
|
||
def get_xray_config_dir(self) -> Path:
|
||
XRAY_CONFIGS_DIR.mkdir(parents=True, exist_ok=True)
|
||
return XRAY_CONFIGS_DIR
|
||
|
||
def get_singbox_template_dir(self) -> Path:
|
||
SINGBOX_TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
|
||
return SINGBOX_TEMPLATES_DIR
|
||
|
||
def get_xray_template_dir(self) -> Path:
|
||
XRAY_TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
|
||
return XRAY_TEMPLATES_DIR
|
||
|
||
def _normalize_singbox_config_relative_path(self, value: str | Path | None) -> str:
|
||
return normalize_relative_json_path(value, SINGBOX_DEFAULT_CONFIG_NAME)
|
||
|
||
def _normalize_singbox_template_relative_path(self, value: str | Path | None) -> str:
|
||
return self._normalize_singbox_config_relative_path(value)
|
||
|
||
def _resolve_singbox_config_path(self, path: str | Path | None = None) -> Path:
|
||
value = self.state.settings.singbox_config_file if path is None or not str(path).strip() else path
|
||
return resolve_profile_path(
|
||
self.get_singbox_config_dir(),
|
||
value,
|
||
SINGBOX_DEFAULT_CONFIG_NAME,
|
||
label="sing-box",
|
||
)
|
||
|
||
def _resolve_singbox_template_path(self, path: str | Path | None = None) -> Path:
|
||
value = self.state.settings.singbox_template_file if path is None or not str(path).strip() else path
|
||
return resolve_profile_path(
|
||
self.get_singbox_template_dir(),
|
||
value,
|
||
SINGBOX_DEFAULT_CONFIG_NAME,
|
||
label="sing-box template",
|
||
)
|
||
|
||
def _normalize_xray_config_relative_path(self, value: str | Path | None) -> str:
|
||
return normalize_relative_json_path(value, XRAY_DEFAULT_CONFIG_NAME)
|
||
|
||
def _normalize_xray_template_relative_path(self, value: str | Path | None) -> str:
|
||
return self._normalize_xray_config_relative_path(value)
|
||
|
||
def _resolve_xray_config_path(self, path: str | Path | None = None) -> Path:
|
||
value = self.state.settings.xray_config_file if path is None or not str(path).strip() else path
|
||
return resolve_profile_path(
|
||
self.get_xray_config_dir(),
|
||
value,
|
||
XRAY_DEFAULT_CONFIG_NAME,
|
||
label="xray",
|
||
)
|
||
|
||
def _resolve_xray_template_path(self, path: str | Path | None = None) -> Path:
|
||
value = self.state.settings.xray_template_file if path is None or not str(path).strip() else path
|
||
return resolve_profile_path(
|
||
self.get_xray_template_dir(),
|
||
value,
|
||
XRAY_DEFAULT_CONFIG_NAME,
|
||
label="xray template",
|
||
)
|
||
|
||
def _set_active_singbox_config_path(self, path: Path, *, emit_signal: bool = True) -> Path:
|
||
resolved = self._resolve_singbox_config_path(path)
|
||
relative = resolved.relative_to(self.get_singbox_config_dir().resolve()).as_posix()
|
||
if self.state.settings.singbox_config_file == relative:
|
||
return resolved
|
||
self.state.settings.singbox_config_file = relative
|
||
if emit_signal:
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.schedule_save()
|
||
return resolved
|
||
|
||
def _set_active_singbox_template_path(self, path: Path, *, emit_signal: bool = True) -> Path:
|
||
resolved = self._resolve_singbox_template_path(path)
|
||
relative = resolved.relative_to(self.get_singbox_template_dir().resolve()).as_posix()
|
||
if self.state.settings.singbox_template_file == relative:
|
||
return resolved
|
||
self.state.settings.singbox_template_file = relative
|
||
if emit_signal:
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.schedule_save()
|
||
return resolved
|
||
|
||
def _set_active_xray_config_path(self, path: Path, *, emit_signal: bool = True) -> Path:
|
||
resolved = self._resolve_xray_config_path(path)
|
||
relative = resolved.relative_to(self.get_xray_config_dir().resolve()).as_posix()
|
||
if self.state.settings.xray_config_file == relative:
|
||
return resolved
|
||
self.state.settings.xray_config_file = relative
|
||
if emit_signal:
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.schedule_save()
|
||
return resolved
|
||
|
||
def _set_active_xray_template_path(self, path: Path, *, emit_signal: bool = True) -> Path:
|
||
resolved = self._resolve_xray_template_path(path)
|
||
relative = resolved.relative_to(self.get_xray_template_dir().resolve()).as_posix()
|
||
if self.state.settings.xray_template_file == relative:
|
||
return resolved
|
||
self.state.settings.xray_template_file = relative
|
||
if emit_signal:
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.schedule_save()
|
||
return resolved
|
||
|
||
@staticmethod
|
||
def _default_singbox_config_text() -> str:
|
||
return default_singbox_config_text()
|
||
|
||
@staticmethod
|
||
def _default_xray_config_text() -> str:
|
||
return default_xray_config_text(
|
||
proxy_host=PROXY_HOST,
|
||
socks_port=DEFAULT_SOCKS_PORT,
|
||
http_port=DEFAULT_HTTP_PORT,
|
||
api_port=DEFAULT_XRAY_STATS_API_PORT,
|
||
)
|
||
|
||
def get_active_singbox_config_path(self) -> Path:
|
||
return get_active_config_path_operation(self, "singbox")
|
||
|
||
def get_active_singbox_config_name(self) -> str:
|
||
return get_active_config_name_operation(self, "singbox")
|
||
|
||
def get_active_singbox_template_path(self) -> Path | None:
|
||
return get_active_template_path_operation(self, "singbox")
|
||
|
||
def get_active_xray_config_path(self) -> Path:
|
||
return get_active_config_path_operation(self, "xray")
|
||
|
||
def get_active_xray_config_name(self) -> str:
|
||
return get_active_config_name_operation(self, "xray")
|
||
|
||
def get_active_xray_template_path(self) -> Path | None:
|
||
return get_active_template_path_operation(self, "xray")
|
||
|
||
def _emit_system_proxy_state(self, *_args) -> None:
|
||
try:
|
||
self.system_proxy_state_changed.emit(self.query_system_proxy_state())
|
||
except Exception: # noqa: BLE001 — снимок для UI не должен ломать переход
|
||
pass
|
||
|
||
def query_system_proxy_state(self) -> SystemProxyState:
|
||
"""Быстрый снимок реального состояния системного прокси Windows.
|
||
|
||
Только чтение реестра (без WinINet/RAS) — безопасно для UI-потока.
|
||
На не-Windows возвращает ``supported=False``.
|
||
"""
|
||
return self.proxy.query_state()
|
||
|
||
def active_tun_mode(self) -> bool | None:
|
||
"""Факт: режим работающей сессии (TUN или прокси); ``None`` — сессии нет.
|
||
|
||
Настройка ``tun_mode`` меняется сразу по клику, а сессия — только после
|
||
перехода; UI сверяет с этим значением выбор пользователя.
|
||
"""
|
||
session = self._active_session
|
||
if session is None or not self.connected:
|
||
return None
|
||
return bool(session.tun_mode)
|
||
|
||
def get_effective_proxy_ports(self) -> tuple[int, int]:
|
||
session = self._active_session
|
||
if session is not None and session.socks_port > 0 and session.http_port > 0:
|
||
return session.socks_port, session.http_port
|
||
if self.is_singbox_proxy_mode():
|
||
return DEFAULT_SOCKS_PORT, DEFAULT_HTTP_PORT
|
||
try:
|
||
_, _, _, socks_port, http_port, _ = self._inspect_active_xray_config()
|
||
except Exception:
|
||
socks_port = 0
|
||
http_port = 0
|
||
if socks_port > 0 and http_port > 0:
|
||
return socks_port, http_port
|
||
return DEFAULT_SOCKS_PORT, DEFAULT_HTTP_PORT
|
||
|
||
def get_effective_http_proxy_port(self) -> int | None:
|
||
session = self._active_session
|
||
if session is not None:
|
||
return session.http_port if session.http_port > 0 else None
|
||
_, http_port = self.get_effective_proxy_ports()
|
||
return http_port if http_port > 0 else None
|
||
|
||
def _cache_singbox_document_state(self, path: Path, text: str) -> SingboxDocumentState:
|
||
return self._singbox_documents.cache_state(path, text)
|
||
|
||
def _persist_singbox_config_repair(
|
||
self,
|
||
path: Path,
|
||
text: str,
|
||
) -> tuple[SingboxDocumentState, SingboxConfigRepair] | None:
|
||
try:
|
||
repair = repair_singbox_config_file(path, text)
|
||
except OSError as exc:
|
||
raise ValueError(
|
||
f"{path.name}: удалось определить безопасное исправление, "
|
||
f"но не удалось сохранить резервную копию или исправленный конфиг: {exc}"
|
||
) from exc
|
||
if repair is None:
|
||
return None
|
||
|
||
repaired_state = self._cache_singbox_document_state(path, repair.repaired_text)
|
||
message = repair.notice(path.name)
|
||
self._log(f"[sing-box] {message}")
|
||
self.status.emit("warning-long", message)
|
||
return repaired_state, repair
|
||
|
||
def _recover_singbox_document_state(self, state: SingboxDocumentState) -> SingboxDocumentState:
|
||
result = self._persist_singbox_config_repair(state.source_path, state.text)
|
||
return result[0] if result is not None else state
|
||
|
||
def _get_singbox_document_state(self) -> SingboxDocumentState:
|
||
path = self._ensure_active_singbox_config()
|
||
state = self._singbox_documents.get_state(path)
|
||
return self._recover_singbox_document_state(state)
|
||
|
||
def _default_singbox_template_path_for_config(self, config_path: Path) -> Path | None:
|
||
relative = config_path.relative_to(self.get_singbox_config_dir().resolve()).as_posix()
|
||
template = self._resolve_singbox_template_path(relative)
|
||
return template if template.exists() else None
|
||
|
||
def _default_xray_template_path_for_config(self, config_path: Path) -> Path | None:
|
||
relative = config_path.relative_to(self.get_xray_config_dir().resolve()).as_posix()
|
||
template = self._resolve_xray_template_path(relative)
|
||
return template if template.exists() else None
|
||
|
||
def _ensure_active_singbox_config(self, path: str | Path | None = None) -> Path:
|
||
return ensure_active_config_operation(self, "singbox", path)
|
||
|
||
def _ensure_active_xray_config(self, path: str | Path | None = None) -> Path:
|
||
return ensure_active_config_operation(self, "xray", path)
|
||
|
||
def load_active_singbox_config_text(self) -> tuple[Path, str]:
|
||
state = self._get_singbox_document_state()
|
||
return state.source_path, state.text
|
||
|
||
def load_active_xray_config_text(self) -> tuple[Path, str]:
|
||
return load_active_config_text_operation(self, "xray")
|
||
|
||
def load_singbox_config_text(self, path: str | Path) -> tuple[Path, str]:
|
||
load_config_text_operation(self, "singbox", path)
|
||
state = self._get_singbox_document_state()
|
||
return state.source_path, state.text
|
||
|
||
def load_xray_config_text(self, path: str | Path) -> tuple[Path, str]:
|
||
return load_config_text_operation(self, "xray", path)
|
||
|
||
def import_singbox_template(self, path: str | Path) -> tuple[Path, str]:
|
||
return import_template_operation(self, "singbox", path)
|
||
|
||
def import_xray_template(self, path: str | Path) -> tuple[Path, str]:
|
||
return import_template_operation(self, "xray", path)
|
||
|
||
def reset_active_singbox_config_to_template(self) -> tuple[bool, Path | None, str]:
|
||
return reset_active_config_to_template_operation(self, "singbox")
|
||
|
||
def reset_active_xray_config_to_template(self) -> tuple[bool, Path | None, str]:
|
||
return reset_active_config_to_template_operation(self, "xray")
|
||
|
||
def save_singbox_config_text(self, text: str, path: str | Path | None = None) -> Path:
|
||
return save_config_text_operation(self, "singbox", text, path)
|
||
|
||
def save_xray_config_text(self, text: str, path: str | Path | None = None) -> Path:
|
||
return save_config_text_operation(self, "xray", text, path)
|
||
|
||
@staticmethod
|
||
def _format_json_error_message(text: str, exc: json.JSONDecodeError) -> str:
|
||
return format_json_error_message(text, exc)
|
||
|
||
def validate_json_text(self, text: str) -> tuple[bool, str]:
|
||
return validate_json_text(text)
|
||
|
||
def validate_singbox_json_text(self, text: str) -> tuple[bool, str]:
|
||
try:
|
||
parse_singbox_document(self.get_active_singbox_config_path(), text)
|
||
except ValueError as exc:
|
||
return False, str(exc)
|
||
return True, "JSON корректен и имеет допустимую структуру конфига sing-box."
|
||
|
||
@staticmethod
|
||
def try_repair_singbox_json_text(text: str) -> SingboxConfigRepair | None:
|
||
return try_repair_singbox_config_text(text)
|
||
|
||
def validate_xray_json_text(self, text: str) -> tuple[bool, str]:
|
||
ok, message = self.validate_json_text(text)
|
||
if not ok:
|
||
return False, message
|
||
if "fakedns" in text.lower():
|
||
return (
|
||
True,
|
||
"JSON корректен. Внимание: в конфиге есть FakeDNS; некоторые версии Xray-core могут падать на старте. "
|
||
"Если запуск завершается с panic, отключите FakeDNS или обновите Xray core.",
|
||
)
|
||
return True, message
|
||
|
||
def apply_singbox_config_text(self, text: str) -> tuple[bool, Path | None, str]:
|
||
return apply_singbox_config_text_operation(self, text)
|
||
|
||
def apply_xray_config_text(self, text: str) -> tuple[bool, Path | None, str]:
|
||
return apply_xray_config_text_operation(self, text)
|
||
|
||
@staticmethod
|
||
def _config_has_proxy_outbound(payload: Any) -> bool:
|
||
return config_has_proxy_outbound(payload)
|
||
|
||
@staticmethod
|
||
def _is_local_runtime_host(value: str) -> bool:
|
||
return is_local_runtime_host(value)
|
||
|
||
@staticmethod
|
||
def _infer_singbox_outbound_endpoint(outbound: dict[str, Any]) -> tuple[str, int]:
|
||
return infer_singbox_outbound_endpoint(outbound)
|
||
|
||
@staticmethod
|
||
def _infer_xray_outbound_endpoint(outbound: dict[str, Any]) -> tuple[str, int]:
|
||
return infer_xray_outbound_endpoint(outbound)
|
||
|
||
@staticmethod
|
||
def _infer_singbox_ping_target(payload: dict[str, Any], node: Node | None) -> tuple[str, int]:
|
||
return infer_singbox_ping_target(payload, node)
|
||
|
||
@staticmethod
|
||
def _infer_xray_ping_target(payload: dict[str, Any], node: Node | None) -> tuple[str, int]:
|
||
return infer_xray_ping_target(payload, node)
|
||
|
||
@staticmethod
|
||
def _ensure_dict(parent: dict[str, Any], key: str) -> dict[str, Any]:
|
||
return ensure_dict(parent, key)
|
||
|
||
@staticmethod
|
||
def _ensure_list(parent: dict[str, Any], key: str) -> list[Any]:
|
||
return ensure_list(parent, key)
|
||
|
||
@staticmethod
|
||
def _replace_or_append_tagged(items: list[Any], tag: str, payload: dict[str, Any]) -> None:
|
||
replace_or_append_tagged(items, tag, payload)
|
||
|
||
@staticmethod
|
||
def _collect_xray_inbound_ports(payload: Any) -> set[int]:
|
||
return collect_xray_inbound_ports(payload)
|
||
|
||
def _ensure_xray_metrics_contract(
|
||
self,
|
||
payload: dict[str, Any],
|
||
*,
|
||
allocate_port: bool,
|
||
) -> tuple[int, tuple[str, ...]]:
|
||
return ensure_xray_metrics_contract_operation(self, payload, allocate_port=allocate_port)
|
||
|
||
def _ensure_xray_tun_contract(self, payload: dict[str, Any]) -> str:
|
||
return ensure_xray_tun_contract_operation(self, payload)
|
||
|
||
@staticmethod
|
||
def _xray_outbound_is_loop_protected(outbound: dict[str, Any]) -> bool:
|
||
return xray_outbound_is_loop_protected_operation(outbound)
|
||
|
||
def _apply_xray_tun_loop_prevention(self, payload: dict[str, Any], interface_alias: str) -> int:
|
||
return apply_xray_tun_loop_prevention_operation(self, payload, interface_alias)
|
||
|
||
def _inspect_active_singbox_config(self) -> tuple[Path, str, bool]:
|
||
state = self._get_singbox_document_state()
|
||
return state.source_path, state.text_hash, state.has_proxy_outbound
|
||
|
||
@staticmethod
|
||
def _extract_xray_runtime_ports(payload: Any) -> tuple[int, int, int]:
|
||
return extract_xray_runtime_ports(payload)
|
||
|
||
def _inspect_active_xray_config(self) -> tuple[Path, str, bool, int, int, int]:
|
||
return inspect_active_xray_config_operation(self)
|
||
|
||
def _plan_runtime_singbox(
|
||
self,
|
||
node: Node | None = None,
|
||
*,
|
||
replacement: bool = False,
|
||
) -> SingboxRuntimePlan:
|
||
state = self._get_singbox_document_state()
|
||
document = parse_singbox_document(state.source_path, state.text)
|
||
preferred_relay_port = 0
|
||
preferred_protect_port = 0
|
||
preferred_protect_password = ""
|
||
allowed_proxy_ports: set[int] = set()
|
||
session = self._active_session
|
||
if session is not None and session.active_core == "singbox":
|
||
# Порты своей же сессии заняты ею самой: переподключение их сохраняет.
|
||
if session.socks_port > 0:
|
||
allowed_proxy_ports.add(int(session.socks_port))
|
||
if session.http_port > 0:
|
||
allowed_proxy_ports.add(int(session.http_port))
|
||
if session is not None and session.active_core == "singbox" and not replacement:
|
||
if session.hybrid:
|
||
preferred_relay_port = session.sidecar_relay_port
|
||
preferred_protect_port = session.protect_ss_port
|
||
preferred_protect_password = session.protect_ss_password
|
||
elif session.sidecar_kind == "hysteria":
|
||
preferred_relay_port = session.sidecar_relay_port
|
||
self._refresh_server_bootstrap(node)
|
||
return plan_singbox_runtime(
|
||
document,
|
||
node,
|
||
allowed_proxy_ports=allowed_proxy_ports,
|
||
preferred_relay_port=preferred_relay_port,
|
||
preferred_protect_port=preferred_protect_port,
|
||
preferred_protect_password=preferred_protect_password,
|
||
pool_nodes=self._pool_candidate_nodes(node),
|
||
)
|
||
|
||
def _refresh_server_bootstrap(self, node: Node | None) -> None:
|
||
"""Обновить в фоне запомненный адрес сервера, не задерживая подключение."""
|
||
|
||
target = node or self.selected_node
|
||
server = str(getattr(target, "server", "") or "")
|
||
if server:
|
||
refresh_bootstrap_address(server)
|
||
|
||
def _plan_proxy_runtime_singbox(
|
||
self,
|
||
node: Node | None = None,
|
||
*,
|
||
replacement: bool = False,
|
||
) -> SingboxRuntimePlan:
|
||
state = self._get_singbox_document_state()
|
||
document = parse_singbox_document(state.source_path, state.text)
|
||
preferred_relay_port = 0
|
||
preferred_protect_port = 0
|
||
preferred_protect_password = ""
|
||
allowed_proxy_ports: set[int] = set()
|
||
session = self._active_session
|
||
if session is not None and session.active_core == "singbox":
|
||
if session.socks_port > 0:
|
||
allowed_proxy_ports.add(int(session.socks_port))
|
||
if session.http_port > 0:
|
||
allowed_proxy_ports.add(int(session.http_port))
|
||
if session.hybrid and not replacement:
|
||
preferred_relay_port = session.sidecar_relay_port
|
||
preferred_protect_port = session.protect_ss_port
|
||
preferred_protect_password = session.protect_ss_password
|
||
elif session.sidecar_kind == "hysteria" and not replacement:
|
||
preferred_relay_port = session.sidecar_relay_port
|
||
self._refresh_server_bootstrap(node)
|
||
return plan_singbox_proxy_runtime(
|
||
document,
|
||
node,
|
||
allowed_proxy_ports=allowed_proxy_ports,
|
||
preferred_relay_port=preferred_relay_port,
|
||
preferred_protect_port=preferred_protect_port,
|
||
preferred_protect_password=preferred_protect_password,
|
||
pool_nodes=self._pool_candidate_nodes(node),
|
||
)
|
||
|
||
def _singbox_plan_facts(self, node: Node | None, *, tun: bool) -> SingboxPlanFacts | None:
|
||
"""Selector tags / provider / used_selected_node for signatures, cached.
|
||
|
||
Сигнатуры переходов и предпроверка выбранного сервера вызывались по
|
||
7–10 раз за клик, и каждый раз строили полный план (outbound'ы всего
|
||
пула, пробы портов). Нужные им поля зависят только от документа
|
||
конфигурации, режима и ноды с составом пула — по этому ключу и кэш.
|
||
Порты из этого кэша не берутся никогда: запуск планирует заново.
|
||
Инвалидация: смена документа (хеш), состава/объектов нод и сигнал
|
||
``nodes_changed`` (in-place правки outbound, инвариант пула).
|
||
"""
|
||
state = self._get_singbox_document_state()
|
||
nodes = self.state.nodes
|
||
self._xray_pool_quarantine_reasons()
|
||
key = (
|
||
str(state.source_path),
|
||
state.text_hash,
|
||
bool(tun),
|
||
self._xray_pool_quarantine_version,
|
||
node.id if node is not None else None,
|
||
id(node.outbound) if node is not None else 0,
|
||
id(nodes),
|
||
tuple((id(item), id(item.outbound), item.sort_order) for item in nodes),
|
||
)
|
||
cache = self._plan_facts_cache
|
||
if key in cache:
|
||
return cache[key]
|
||
try:
|
||
plan = (
|
||
self._plan_runtime_singbox(node)
|
||
if tun
|
||
else self._plan_proxy_runtime_singbox(node)
|
||
)
|
||
facts: SingboxPlanFacts | None = SingboxPlanFacts(
|
||
selector_tags=dict(plan.selector_tags or {}),
|
||
provider_payload=plan.provider_payload,
|
||
used_selected_node=bool(plan.used_selected_node),
|
||
)
|
||
except ValueError:
|
||
facts = None
|
||
if len(cache) >= 8:
|
||
cache.pop(next(iter(cache)))
|
||
cache[key] = facts
|
||
return facts
|
||
|
||
def _invalidate_plan_facts_cache(self) -> None:
|
||
self._plan_facts_cache.clear()
|
||
|
||
def _configure_singbox_log_contexts(self, plan: SingboxRuntimePlan) -> None:
|
||
"""Freeze tag-to-node mappings before any runtime process starts."""
|
||
|
||
mode = "tun" if self.state.settings.tun_mode else "proxy"
|
||
generation = int(self._transition_generation)
|
||
nodes_by_id = {node.id: node for node in self.state.nodes}
|
||
selected_node = self._runtime_selected_node() if plan.used_selected_node else None
|
||
selected = RuntimeNodeIdentity.from_node(selected_node) if selected_node is not None else None
|
||
|
||
singbox_nodes = identities_for_tags(plan.selector_tags, nodes_by_id)
|
||
if selected is not None:
|
||
singbox_nodes.setdefault("proxy", selected)
|
||
singbox_context = RuntimeLogContext(
|
||
engine="sing-box",
|
||
role="front",
|
||
mode=mode,
|
||
generation=generation,
|
||
selected=selected,
|
||
outbound_nodes=singbox_nodes,
|
||
)
|
||
self._core_log_contexts["sing-box"] = singbox_context
|
||
|
||
if plan.hysteria_sidecar is not None:
|
||
self._core_log_contexts["hysteria"] = RuntimeLogContext(
|
||
engine="hysteria",
|
||
role="sidecar",
|
||
mode=mode,
|
||
generation=generation,
|
||
selected=plan.hysteria_sidecar.context,
|
||
outbound_nodes={"proxy": plan.hysteria_sidecar.context},
|
||
)
|
||
else:
|
||
self._core_log_contexts.pop("hysteria", None)
|
||
|
||
if plan.xray_sidecar is not None:
|
||
xray_nodes = identities_for_tags(plan.xray_sidecar.outbound_pool_tags, nodes_by_id)
|
||
if selected is not None:
|
||
xray_nodes.setdefault("proxy", selected)
|
||
self._core_log_contexts["xray"] = RuntimeLogContext(
|
||
engine="xray",
|
||
role="sidecar",
|
||
mode=mode,
|
||
generation=generation,
|
||
selected=selected,
|
||
outbound_nodes=xray_nodes,
|
||
)
|
||
else:
|
||
self._core_log_contexts.pop("xray", None)
|
||
|
||
for context in (
|
||
singbox_context,
|
||
self._core_log_contexts.get("hysteria"),
|
||
self._core_log_contexts.get("xray") if plan.xray_sidecar is not None else None,
|
||
):
|
||
if context is None:
|
||
continue
|
||
for line in runtime_mapping_lines(context):
|
||
self._log(line)
|
||
|
||
def _configure_core_log_context(
|
||
self,
|
||
engine: str,
|
||
*,
|
||
node: Node | None,
|
||
outbound_tags: dict[str, str] | None = None,
|
||
role: str = "core",
|
||
used_selected_node: bool = True,
|
||
) -> None:
|
||
"""Freeze a secret-safe node mapping before starting a standalone core."""
|
||
|
||
selected_node = node if used_selected_node else None
|
||
selected = RuntimeNodeIdentity.from_node(selected_node) if selected_node is not None else None
|
||
nodes_by_id = {item.id: item for item in self.state.nodes}
|
||
mapped = identities_for_tags(outbound_tags or {}, nodes_by_id)
|
||
if selected is not None:
|
||
mapped.setdefault("proxy", selected)
|
||
context = RuntimeLogContext(
|
||
engine=engine,
|
||
role=role,
|
||
mode="tun" if self.state.settings.tun_mode else "proxy",
|
||
generation=int(self._transition_generation),
|
||
selected=selected,
|
||
outbound_nodes=mapped,
|
||
)
|
||
self._core_log_contexts[engine] = context
|
||
for line in runtime_mapping_lines(context):
|
||
self._log(line)
|
||
|
||
def _start_singbox_runtime_plan(
|
||
self,
|
||
plan: SingboxRuntimePlan,
|
||
*,
|
||
prepared_hysteria: HysteriaManager | None = None,
|
||
prepared_amnezia: AmneziaManager | None = None,
|
||
) -> bool:
|
||
"""Синхронный драйвер (тесты). Переходы — ``_start_singbox_runtime_plan_steps``."""
|
||
return bool(
|
||
run_steps_blocking(
|
||
self._start_singbox_runtime_plan_steps(
|
||
plan,
|
||
prepared_hysteria=prepared_hysteria,
|
||
prepared_amnezia=prepared_amnezia,
|
||
)
|
||
)
|
||
)
|
||
|
||
def _start_singbox_runtime_plan_steps(
|
||
self,
|
||
plan: SingboxRuntimePlan,
|
||
*,
|
||
prepared_hysteria: HysteriaManager | None = None,
|
||
prepared_amnezia: AmneziaManager | None = None,
|
||
) -> TransitionSteps:
|
||
request_generation = getattr(self, "_transition_generation", None)
|
||
owned_amnezia = prepared_amnezia or getattr(self, "amnezia", None)
|
||
|
||
def request_cancelled() -> bool:
|
||
return (
|
||
getattr(self, "_desired_connected", True) is False or
|
||
getattr(self, "_transition_generation", None) != request_generation
|
||
)
|
||
|
||
if request_cancelled():
|
||
return False
|
||
runtime_node = self._runtime_selected_node()
|
||
gate = getattr(self, "target_profile_allows_core_start", None)
|
||
if gate is not None and not gate(
|
||
runtime_node,
|
||
used_selected_node=bool(getattr(plan, "used_selected_node", True)),
|
||
):
|
||
return False
|
||
self._configure_singbox_log_contexts(plan)
|
||
if plan.provider_payload is not None:
|
||
# Провайдер несёт outbound'ы всего пула: сериализация и запись — в воркере.
|
||
provider_payload = plan.provider_payload
|
||
provider_file = SINGBOX_PROVIDER_FILE
|
||
yield run_in_worker(lambda: _write_json_atomically(provider_file, provider_payload))
|
||
if request_cancelled():
|
||
return False
|
||
should_continue = lambda: not request_cancelled() # noqa: E731
|
||
if getattr(plan, "amnezia_sidecar", None) is not None:
|
||
target_amnezia = owned_amnezia
|
||
if prepared_amnezia is not None:
|
||
if not target_amnezia.is_running:
|
||
return False
|
||
elif not (yield from self._start_amnezia_manager_steps(target_amnezia, plan)):
|
||
return False
|
||
if request_cancelled():
|
||
yield from target_amnezia.stop_steps()
|
||
return False
|
||
if plan.hysteria_sidecar is not None:
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
self._log(
|
||
"[sing-box] starting official Hysteria2 sidecar "
|
||
f"relay=127.0.0.1:{plan.hysteria_sidecar.relay_port}"
|
||
)
|
||
target_manager = prepared_hysteria or self.hysteria
|
||
if prepared_hysteria is not None:
|
||
if not prepared_hysteria.is_running:
|
||
return False
|
||
else:
|
||
contract = getattr(self, "_hysteria_contract", None)
|
||
if isinstance(contract, HysteriaTransitionContract):
|
||
planned_generation = _increment_int(
|
||
getattr(self, "_session_generation", 0)
|
||
)
|
||
contract.begin(
|
||
planned_generation,
|
||
runtime_node.id if runtime_node else None,
|
||
"official_hysteria_sidecar",
|
||
)
|
||
contract.advance(
|
||
HysteriaRuntimeState.STARTING_SIDECAR,
|
||
generation=planned_generation,
|
||
)
|
||
contract.advance(
|
||
HysteriaRuntimeState.WAITING_RELAY,
|
||
generation=planned_generation,
|
||
)
|
||
self._hysteria_process_generation = _increment_int(
|
||
getattr(self, "_hysteria_process_generation", 0)
|
||
)
|
||
self._hysteria_active_generation = self._hysteria_process_generation
|
||
if not (yield from target_manager.start_steps(
|
||
plan.hysteria_sidecar.config,
|
||
plan.hysteria_sidecar.relay_port,
|
||
context=plan.hysteria_sidecar.context,
|
||
process_generation=self._hysteria_process_generation,
|
||
should_continue=should_continue,
|
||
)):
|
||
return False
|
||
elif plan.xray_sidecar is not None:
|
||
self._protect_ss_port = plan.xray_sidecar.protect_port
|
||
self._protect_ss_password = plan.xray_sidecar.protect_password
|
||
self._log(
|
||
"[sing-box] starting hybrid xray sidecar "
|
||
f"relay=127.0.0.1:{plan.xray_sidecar.relay_port} "
|
||
f"protect=127.0.0.1:{plan.xray_sidecar.protect_port}"
|
||
)
|
||
if not (yield from self.xray.start_steps(self.state.settings.xray_path, plan.xray_sidecar.config)):
|
||
# Что ядро отвергло, не возвращается в следующий запуск: он должен сойтись.
|
||
self._quarantine_pruned_xray_outbounds(plan, self._rejected_xray_tags(), notify=False)
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
return False
|
||
pruned_tags = self._rejected_xray_tags()
|
||
if pruned_tags:
|
||
rejected_selected = self._quarantine_pruned_xray_outbounds(plan, pruned_tags)
|
||
if rejected_selected:
|
||
yield from self.xray.stop_steps()
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
self.xray.error.emit(rejected_selected)
|
||
return False
|
||
self._xray_api_port = plan.xray_sidecar.api_port
|
||
if plan.selected_outbound_tag and not (yield from self._apply_core_outbound_tag_steps(
|
||
"xray", plan.selected_outbound_tag
|
||
)):
|
||
yield from self.xray.stop_steps()
|
||
self._xray_api_port = 0
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
return False
|
||
else:
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
|
||
if plan.clash_api_port <= 0:
|
||
self._log(
|
||
"[sing-box] clash_api отключён: свободный порт метрик не найден "
|
||
f"(диапазон {SINGBOX_CLASH_API_PORT}+ зарезервирован Windows)"
|
||
)
|
||
elif plan.clash_api_port != SINGBOX_CLASH_API_PORT:
|
||
self._log(
|
||
f"[sing-box] clash_api порт изменён: {SINGBOX_CLASH_API_PORT} -> {plan.clash_api_port} "
|
||
"(исходный зарезервирован Windows)"
|
||
)
|
||
if request_cancelled() or (gate is not None and not gate(
|
||
runtime_node,
|
||
used_selected_node=bool(getattr(plan, "used_selected_node", True)),
|
||
)):
|
||
if plan.xray_sidecar is not None and self.xray.is_running:
|
||
yield from self.xray.stop_steps()
|
||
if plan.hysteria_sidecar is not None:
|
||
target_manager = prepared_hysteria or self.hysteria
|
||
if target_manager.is_running:
|
||
yield from target_manager.stop_steps()
|
||
if getattr(plan, "amnezia_sidecar", None) is not None:
|
||
yield from (prepared_amnezia or self.amnezia).stop_steps()
|
||
return False
|
||
singbox_start_tag = (
|
||
plan.hybrid_relay_selected_tag if plan.is_hybrid else plan.selected_outbound_tag
|
||
)
|
||
if plan.hysteria_sidecar is not None:
|
||
contract = getattr(self, "_hysteria_contract", None)
|
||
if isinstance(contract, HysteriaTransitionContract):
|
||
contract.advance(
|
||
HysteriaRuntimeState.STARTING_FRONT,
|
||
generation=contract.session.session_generation,
|
||
)
|
||
for start_attempt in range(2):
|
||
if request_cancelled():
|
||
break
|
||
sb_ok = yield from self.singbox.start_steps(
|
||
self.state.settings.singbox_path,
|
||
plan.singbox_config,
|
||
should_continue=should_continue,
|
||
)
|
||
self._log(f"[sing-box] start result: {sb_ok}")
|
||
if request_cancelled():
|
||
yield from self.singbox.stop_steps()
|
||
break
|
||
if not sb_ok:
|
||
retryable = getattr(
|
||
self.singbox,
|
||
"last_start_failure_retryable",
|
||
False,
|
||
) is True
|
||
if start_attempt == 0 and retryable:
|
||
self._log(
|
||
"[sing-box] self-heal: control plane did not become ready; "
|
||
"restarting sing-box once"
|
||
)
|
||
yield sleep_ms(500)
|
||
continue
|
||
break
|
||
self._singbox_clash_api_port = plan.clash_api_port
|
||
if not singbox_start_tag or (yield from self._apply_core_outbound_tag_steps(
|
||
"singbox", singbox_start_tag, startup=True
|
||
)):
|
||
if getattr(plan, "amnezia_sidecar", None) is not None:
|
||
if not owned_amnezia.verify_front_dns(plan.amnezia_sidecar.config) or request_cancelled():
|
||
yield from self.singbox.stop_steps()
|
||
break
|
||
if request_cancelled():
|
||
yield from self.singbox.stop_steps()
|
||
break
|
||
self._front_process_generation = _increment_int(
|
||
getattr(self, "_front_process_generation", 0)
|
||
)
|
||
self._front_target_generation = _increment_int(
|
||
getattr(self, "_front_target_generation", 0)
|
||
)
|
||
self._active_singbox_plan = plan
|
||
return True
|
||
|
||
yield from self.singbox.stop_steps()
|
||
self._singbox_clash_api_port = 0
|
||
if start_attempt == 0 and not request_cancelled():
|
||
self._log(
|
||
"[sing-box] self-heal: control plane did not become ready; "
|
||
"restarting sing-box once"
|
||
)
|
||
yield sleep_ms(500)
|
||
|
||
if plan.xray_sidecar is not None and (self.xray.is_running or _process_alive(self.xray)):
|
||
yield from self.xray.stop_steps()
|
||
if getattr(plan, "amnezia_sidecar", None) is not None:
|
||
yield from owned_amnezia.stop_steps()
|
||
if plan.hysteria_sidecar is not None:
|
||
target_manager = prepared_hysteria or self.hysteria
|
||
if target_manager.is_running or _process_alive(target_manager):
|
||
yield from target_manager.stop_steps()
|
||
self._protect_ss_port = 0
|
||
self._protect_ss_password = ""
|
||
self._singbox_clash_api_port = 0
|
||
self._xray_api_port = 0
|
||
return False
|
||
|
||
def _new_amnezia_manager(self) -> AmneziaManager:
|
||
manager = AmneziaManager(self)
|
||
manager.log_received.connect(self._log)
|
||
manager.warning.connect(lambda message, owned=manager: self._on_amnezia_warning(owned, message))
|
||
manager.failure.connect(lambda event, owned=manager: self._on_amnezia_failure(owned, event))
|
||
return manager
|
||
|
||
def _on_amnezia_warning(self, manager, message) -> None:
|
||
if manager is self.amnezia and manager.is_running:
|
||
self.status.emit("warning", message)
|
||
|
||
def _on_amnezia_failure(self, manager, event) -> None:
|
||
self.runtime_errors.record(event)
|
||
self.runtime_errors_changed.emit(self.runtime_errors.snapshot())
|
||
# A startup failure of the live manager means the AWG sidecar we just
|
||
# tried to bring up never started. Surface it even when no amnezia
|
||
# session is committed yet (fresh connect, or a switch from another
|
||
# protocol) — otherwise AWG "just exits" with nothing in the info bar.
|
||
# Retired replacement candidates are not self.amnezia, so their noise
|
||
# is still filtered.
|
||
if manager is self.amnezia and event.stage == "startup":
|
||
self._set_connection_status("error", event.message, level="error")
|
||
return
|
||
# Retain old/candidate evidence, but only the committed manager owns
|
||
# admission. Never let a retired process close the replacement.
|
||
session = self._active_session
|
||
if manager is not self.amnezia or session is None or session.sidecar_kind != "amnezia":
|
||
return
|
||
if event.stage in {"destination_dns", "relay_connection", "udp", "handshake_retry"}:
|
||
# A failed destination is evidence, not a failure of the whole VPN.
|
||
# The bounded health check reports its aggregate warning separately.
|
||
return
|
||
self._set_connection_status("error", event.message, level="error")
|
||
if event.stage in {"core_error", "process", "observer", "handshake_failed"}:
|
||
def close_owned():
|
||
if manager is self.amnezia and self._active_session is session:
|
||
self._desired_connected = False
|
||
self._handle_unexpected_disconnect()
|
||
QTimer.singleShot(0, close_owned)
|
||
|
||
def _start_amnezia_manager_steps(self, manager, plan: SingboxRuntimePlan) -> TransitionSteps:
|
||
sidecar = plan.amnezia_sidecar
|
||
if sidecar is None:
|
||
return False
|
||
self._amnezia_target_generation = _increment_int(getattr(self, "_amnezia_target_generation", 0))
|
||
generation = self._transition_generation
|
||
return (yield from manager.start_steps(
|
||
sidecar.config, sidecar.relay_port, context=sidecar.context,
|
||
session_generation=_increment_int(getattr(self, "_session_generation", 0)),
|
||
target_generation=self._amnezia_target_generation,
|
||
is_current=lambda: self._desired_connected and self._transition_generation == generation,
|
||
))
|
||
|
||
def _prepare_amnezia_replacement_steps(self, plan: SingboxRuntimePlan) -> TransitionSteps:
|
||
candidate = self._new_amnezia_manager()
|
||
if (yield from self._start_amnezia_manager_steps(candidate, plan)):
|
||
return candidate
|
||
yield from candidate.stop_steps()
|
||
candidate.deleteLater()
|
||
return None
|
||
|
||
def _prepare_hysteria_replacement_steps(self, plan: SingboxRuntimePlan) -> TransitionSteps:
|
||
sidecar = plan.hysteria_sidecar
|
||
if sidecar is None:
|
||
return None
|
||
self._hysteria_process_generation = _increment_int(
|
||
getattr(self, "_hysteria_process_generation", 0)
|
||
)
|
||
generation = self._hysteria_process_generation
|
||
replacement = HysteriaManager(self)
|
||
self._bind_hysteria_manager(replacement)
|
||
self._hysteria_contract.advance(
|
||
HysteriaRuntimeState.STARTING_SIDECAR,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
self._hysteria_contract.advance(
|
||
HysteriaRuntimeState.WAITING_RELAY,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
request_generation = self._transition_generation
|
||
if not (yield from replacement.start_steps(
|
||
sidecar.config,
|
||
sidecar.relay_port,
|
||
context=sidecar.context,
|
||
process_generation=generation,
|
||
allow_parallel=True,
|
||
should_continue=lambda: (
|
||
self._desired_connected and self._transition_generation == request_generation
|
||
),
|
||
)):
|
||
failure = replacement.last_failure_code or HysteriaFailureCode.LOCAL_RELAY_NOT_READY
|
||
self._hysteria_last_failure_code = failure
|
||
self._hysteria_contract.terminal(
|
||
failure,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
replacement.deleteLater()
|
||
return None
|
||
self._relay_credentials_generation = _increment_int(
|
||
getattr(self, "_relay_credentials_generation", 0)
|
||
)
|
||
self._hysteria_contract.session.sidecar_process_generation = generation
|
||
self._hysteria_contract.session.relay_port = sidecar.relay_port
|
||
self._hysteria_contract.session.relay_credentials_generation = self._relay_credentials_generation
|
||
self._hysteria_contract.advance(
|
||
HysteriaRuntimeState.REPLACEMENT_READY,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
return replacement
|
||
|
||
def _commit_hysteria_replacement_steps(self, replacement: HysteriaManager | None) -> TransitionSteps:
|
||
old = self.hysteria
|
||
if replacement is not None:
|
||
self.hysteria = replacement
|
||
self._hysteria_active_generation = replacement.process_generation
|
||
stopped = True
|
||
if old is not replacement and old.is_running:
|
||
stopped = yield from old.stop_steps(expected=True)
|
||
if old is not replacement and stopped:
|
||
old.deleteLater()
|
||
elif old is not replacement:
|
||
old.stopped.connect(old.deleteLater)
|
||
return stopped
|
||
|
||
def _rollback_singbox_front_steps(self, plan: SingboxRuntimePlan | None) -> TransitionSteps:
|
||
if plan is None:
|
||
return False
|
||
if plan.hysteria_sidecar is not None and not self.hysteria.is_running:
|
||
return False
|
||
if getattr(plan, "amnezia_sidecar", None) is not None and not self.amnezia.is_running:
|
||
return False
|
||
if plan.xray_sidecar is not None and not self.xray.is_running:
|
||
return False
|
||
self._log("[transport-transition] replacement front failed; restoring previous generation")
|
||
if not (yield from self.singbox.start_steps(self.state.settings.singbox_path, plan.singbox_config)):
|
||
return False
|
||
self._singbox_clash_api_port = plan.clash_api_port
|
||
selected = plan.hybrid_relay_selected_tag if plan.is_hybrid else plan.selected_outbound_tag
|
||
if selected and not (yield from self._apply_core_outbound_tag_steps("singbox", selected, startup=True)):
|
||
yield from self.singbox.stop_steps()
|
||
return False
|
||
self._active_singbox_plan = plan
|
||
return True
|
||
|
||
def _bind_hysteria_manager(self, manager: HysteriaManager) -> None:
|
||
manager.warning.connect(lambda message, current=manager: self._on_hysteria_warning(current, message))
|
||
manager.log_received.connect(lambda line: self._on_core_log("hysteria", line))
|
||
manager.error.connect(lambda message: self._on_core_error("hysteria", message))
|
||
manager.failure.connect(
|
||
lambda code, message, generation, current=manager: self._on_hysteria_failure(
|
||
current,
|
||
code,
|
||
message,
|
||
generation,
|
||
)
|
||
)
|
||
manager.state_changed.connect(
|
||
lambda running, current=manager: self._on_hysteria_state_changed(current, running)
|
||
)
|
||
manager.stopped.connect(
|
||
lambda code, current=manager: self._on_hysteria_stopped(current, code)
|
||
)
|
||
|
||
def _on_hysteria_warning(self, manager: HysteriaManager, message: str) -> None:
|
||
if manager is self.hysteria and manager.process_generation == self._hysteria_active_generation:
|
||
self.status.emit("warning", message)
|
||
|
||
def _on_hysteria_state_changed(self, manager: HysteriaManager, running: bool) -> None:
|
||
if manager is not self.hysteria or manager.process_generation != self._hysteria_active_generation:
|
||
self._log(
|
||
"[hysteria-transition] ignored stale state callback "
|
||
f"generation={manager.process_generation} running={running}"
|
||
)
|
||
return
|
||
self._on_core_state_changed(running)
|
||
|
||
def _on_hysteria_stopped(self, manager: HysteriaManager, exit_code: int) -> None:
|
||
suffix = ""
|
||
if manager.last_failure_code is not None:
|
||
suffix = f" original_failure={manager.last_failure_code.value}"
|
||
self._log(
|
||
f"[hysteria] process stopped with code {exit_code} "
|
||
f"generation={manager.process_generation}{suffix}"
|
||
)
|
||
|
||
def _on_hysteria_failure(
|
||
self,
|
||
manager: HysteriaManager,
|
||
code_text: str,
|
||
message: str,
|
||
process_generation: int,
|
||
) -> None:
|
||
try:
|
||
code = HysteriaFailureCode(code_text)
|
||
except ValueError:
|
||
code = HysteriaFailureCode.CORE_UNCLASSIFIED
|
||
self._log(
|
||
f"[hysteria-failure] code={code.value} process_generation={process_generation}"
|
||
)
|
||
if manager is not self.hysteria or process_generation != self._hysteria_active_generation:
|
||
self._log("[hysteria-failure] stale generation ignored")
|
||
return
|
||
self._hysteria_last_failure_code = code
|
||
if code in SECURITY_FAILURES and self.connected and not self._disconnecting:
|
||
if getattr(self, "_hysteria_recovery_active", False):
|
||
self._hysteria_contract.terminal(
|
||
code,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
# Отказал резервный сервер, а не выбранный пользователем:
|
||
# его надолго исключаем и возвращаемся на выбранный.
|
||
self._reject_hysteria_replacement(3600.0, f"replacement rejected ({code.value})")
|
||
else:
|
||
self._hysteria_contract.fail(
|
||
code,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
automatic_switch=False,
|
||
)
|
||
self._set_connection_status(
|
||
"error",
|
||
message,
|
||
level="error",
|
||
)
|
||
self._desired_connected = False
|
||
# Fail closed: do not leave the front admitting connections into a
|
||
# transport whose pin/CA/auth/obfs contract was rejected.
|
||
if self.singbox.is_running:
|
||
self.singbox.request_stop(expected=True)
|
||
return
|
||
if (
|
||
code in AUTOMATIC_SWITCH_FAILURES
|
||
and self.connected
|
||
and self._hysteria_recovery_active
|
||
and not self._disconnecting
|
||
):
|
||
self._reject_hysteria_replacement(300.0, f"replacement failed ({code.value})")
|
||
return
|
||
if (
|
||
code not in AUTOMATIC_SWITCH_FAILURES
|
||
or not self.connected
|
||
or self._hysteria_recovery_active
|
||
or self._disconnecting
|
||
or not self._desired_connected
|
||
):
|
||
return
|
||
|
||
failed = self.selected_node
|
||
failed_id = failed.id if failed is not None else ""
|
||
now = time.monotonic()
|
||
self._hysteria_failure_started_at = now
|
||
if failed_id:
|
||
self._hysteria_cooldown_until[failed_id] = now + 300.0
|
||
|
||
replacement: Node | None = None
|
||
for candidate in sorted(self.state.nodes, key=lambda item: item.sort_order):
|
||
if candidate.id == failed_id or node_is_maintenance(candidate):
|
||
continue
|
||
if self._hysteria_cooldown_until.get(candidate.id, 0.0) > now:
|
||
continue
|
||
if getattr(candidate, "is_alive", None) is False:
|
||
continue
|
||
raw_uri = str(candidate.link or "")
|
||
if raw_uri.partition(":")[0].lower() in {"hy2", "hysteria2"}:
|
||
if not classify_hysteria_uri(raw_uri, platform="windows").valid:
|
||
continue
|
||
elif classify_node_for_singbox(candidate) not in {
|
||
"native_singbox",
|
||
"native_singbox_endpoint",
|
||
"hybrid_xray_sidecar",
|
||
}:
|
||
continue
|
||
replacement = candidate
|
||
break
|
||
|
||
self._hysteria_failure_episode_id += 1
|
||
self._hysteria_automatic_switch_attempted = replacement is not None
|
||
self._hysteria_contract.fail(
|
||
code,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
automatic_switch=replacement is not None,
|
||
)
|
||
if replacement is None:
|
||
self._hysteria_last_failure_code = HysteriaFailureCode.NO_COMPATIBLE_FALLBACK
|
||
self._hysteria_contract.terminal(
|
||
HysteriaFailureCode.NO_COMPATIBLE_FALLBACK,
|
||
generation=self._hysteria_contract.session.session_generation,
|
||
)
|
||
self._set_connection_status(
|
||
"error",
|
||
"Hysteria2: совместимый резервный сервер не найден.",
|
||
level="error",
|
||
)
|
||
# Stop admission immediately; otherwise front keeps opening new
|
||
# connections against an already dead loopback relay.
|
||
if self.singbox.is_running:
|
||
self.singbox.request_stop(expected=True)
|
||
self._desired_connected = False
|
||
self._schedule_return_to_selected("no replacement server available")
|
||
return
|
||
|
||
self._hysteria_recovery_active = True
|
||
log_selection_change(
|
||
self, SelectionSource.HYSTERIA_RECOVERY, selection_snapshot(self, failed_id or None), replacement.id
|
||
)
|
||
self._pending_transport_node_id = replacement.id
|
||
self._auto_switch_transitioning = True
|
||
self._switching = True
|
||
self._desired_connected = True
|
||
self._log(
|
||
f"[hysteria-recovery] episode={self._hysteria_failure_episode_id} "
|
||
f"failed={failed_id} replacement={replacement.id} code={code.value}"
|
||
)
|
||
self.auto_switch_triggered.emit(replacement.name)
|
||
# The committed selection remains authoritative until replacement
|
||
# sidecar readiness, front readiness and active-session capture all
|
||
# succeed. Transition planners read the pending node explicitly.
|
||
self._request_transition("node switched")
|
||
# Close admission to the failed loopback relay once. The recovery
|
||
# state fence above keeps the logical old session available to the
|
||
# full-transition planner while no connection-refused storm is created.
|
||
if self.singbox.is_running:
|
||
# kill без ожидания: выход процесса придёт сигналом, а переход
|
||
# дождётся его шагом перед запуском нового front'а.
|
||
self.singbox.request_stop(expected=True)
|
||
|
||
def _reject_hysteria_replacement(self, cooldown_sec: float, reason: str) -> None:
|
||
replacement_id = self._pending_transport_node_id
|
||
if isinstance(replacement_id, str) and replacement_id:
|
||
self._hysteria_cooldown_until[replacement_id] = time.monotonic() + cooldown_sec
|
||
self._schedule_return_to_selected(reason)
|
||
|
||
def _schedule_return_to_selected(self, reason: str) -> None:
|
||
"""Замена не удалась: не оставаться отключённым, вернуться на свой сервер.
|
||
|
||
Возврат — обычное подключение к выбранному серверу после того, как
|
||
неудачный переход полностью завершится; один раз на эпизод отказа.
|
||
"""
|
||
|
||
episode = self._hysteria_failure_episode_id
|
||
self._hysteria_return_episode = episode
|
||
self._log(f"[hysteria-recovery] {reason}; returning to the selected server")
|
||
QTimer.singleShot(2000, lambda: self._return_to_selected_server(episode, 0))
|
||
|
||
def _return_to_selected_server(self, episode: int, attempt: int) -> None:
|
||
if self._hysteria_return_episode != episode or self.locked:
|
||
return
|
||
if self._transition_active or self._transition_pending or self._disconnecting:
|
||
if attempt < 10:
|
||
QTimer.singleShot(
|
||
1000, lambda: self._return_to_selected_server(episode, attempt + 1)
|
||
)
|
||
return
|
||
self._hysteria_return_episode = 0
|
||
if self.connected or self._desired_connected or self.selected_node is None:
|
||
return
|
||
self._log("[hysteria-recovery] reconnecting to the selected server")
|
||
self._desired_connected = True
|
||
self._request_transition("return to selected server")
|
||
|
||
def _record_hysteria_switch_commit(self) -> None:
|
||
started = float(getattr(self, "_hysteria_failure_started_at", 0.0))
|
||
if started <= 0:
|
||
return
|
||
elapsed_ms = int(max(0.0, time.monotonic() - started) * 1000)
|
||
self._log(f"[hysteria-recovery] switch committed in {elapsed_ms} ms")
|
||
self._hysteria_failure_started_at = 0.0
|
||
|
||
def _build_runtime_xray_config(self, node: Node | None = None, *, tun_mode: bool = False) -> XrayRuntimeConfig:
|
||
return build_runtime_xray_config_operation(self, node, tun_mode=tun_mode)
|
||
|
||
def _transition_signature(
|
||
self,
|
||
node: Node | None = None,
|
||
settings: AppSettings | None = None,
|
||
routing: RoutingSettings | None = None,
|
||
) -> str:
|
||
return transition_signature_operation(self, node, settings, routing)
|
||
|
||
def _xray_layer_signature(
|
||
self,
|
||
node: Node | None = None,
|
||
settings: AppSettings | None = None,
|
||
routing: RoutingSettings | None = None,
|
||
) -> str:
|
||
return xray_layer_signature_operation(self, node, settings, routing)
|
||
|
||
def _tun_layer_signature(
|
||
self,
|
||
node: Node | None = None,
|
||
settings: AppSettings | None = None,
|
||
routing: RoutingSettings | None = None,
|
||
) -> str:
|
||
return tun_layer_signature_operation(self, node, settings, routing)
|
||
|
||
def _capture_active_session(
|
||
self,
|
||
node: Node | None,
|
||
*,
|
||
tun: bool,
|
||
core: str,
|
||
api_port: int,
|
||
hybrid: bool = False,
|
||
socks_port: int | None = None,
|
||
http_port: int | None = None,
|
||
xray_inbound_tags: tuple[str, ...] | None = None,
|
||
sidecar_relay_port: int = 0,
|
||
protect_ss_port: int = 0,
|
||
protect_ss_password: str = "",
|
||
ping_host: str = "",
|
||
ping_port: int = 0,
|
||
outbound_pool_tags: dict[str, str] | None = None,
|
||
hybrid_relay_selector_tags: tuple[str, ...] = (),
|
||
hybrid_relay_selected_tag: str = "",
|
||
sidecar_kind: str = "",
|
||
) -> None:
|
||
settings = self.state.settings
|
||
routing = self.state.routing
|
||
if socks_port is None:
|
||
socks_port = int(DEFAULT_SOCKS_PORT)
|
||
if http_port is None:
|
||
http_port = int(DEFAULT_HTTP_PORT)
|
||
if xray_inbound_tags is None:
|
||
xray_inbound_tags = ()
|
||
if not ping_host and node is not None:
|
||
ping_host = node.server
|
||
if ping_port <= 0 and node is not None:
|
||
ping_port = int(node.port)
|
||
proxy_bypass_lan = bool(routing.bypass_lan) if tun else self._system_proxy_bypass_lan(settings)
|
||
if outbound_pool_tags is None:
|
||
# П1 (AC1): самодостаточный дефолт — теги выводятся из единого
|
||
# источника (пула контроллера), а не из памяти вызывающего.
|
||
# Явно переданный параметр (в т.ч. пустой dict) всегда сильнее.
|
||
outbound_pool_tags = self._derive_outbound_pool_tags(node, core=core)
|
||
self._session_generation = _increment_int(getattr(self, "_session_generation", 0))
|
||
contract = getattr(self, "_hysteria_contract", None)
|
||
contract_session = getattr(contract, "session", None)
|
||
started_at = (
|
||
float(getattr(contract_session, "started_at_monotonic", 0.0))
|
||
or time.monotonic()
|
||
)
|
||
ready_at = time.monotonic()
|
||
runtime_kind = (
|
||
"official_hysteria_sidecar"
|
||
if sidecar_kind == "hysteria"
|
||
else "native"
|
||
)
|
||
self._active_session = build_active_session_snapshot(
|
||
node_id=node.id if node else None,
|
||
node_server=node.server if node else "",
|
||
active_core=core,
|
||
tun_mode=bool(tun),
|
||
tun_engine=str(settings.tun_engine),
|
||
proxy_enabled=bool(settings.enable_system_proxy),
|
||
proxy_bypass_lan=proxy_bypass_lan,
|
||
xray_path=str(settings.xray_path),
|
||
singbox_path=str(settings.singbox_path),
|
||
socks_port=int(socks_port),
|
||
http_port=int(http_port),
|
||
routing_signature=self._routing_signature(routing),
|
||
transition_signature=self._transition_signature(node, settings, routing),
|
||
xray_layer_signature=self._xray_layer_signature(node, settings, routing),
|
||
tun_layer_signature=self._tun_layer_signature(node, settings, routing),
|
||
hybrid=hybrid,
|
||
api_port=int(api_port),
|
||
xray_inbound_tags=tuple(xray_inbound_tags),
|
||
sidecar_relay_port=int(sidecar_relay_port),
|
||
protect_ss_port=int(protect_ss_port),
|
||
protect_ss_password=str(protect_ss_password),
|
||
ping_host=str(ping_host),
|
||
ping_port=int(ping_port),
|
||
outbound_pool_tags=outbound_pool_tags,
|
||
hybrid_relay_selector_tags=hybrid_relay_selector_tags,
|
||
hybrid_relay_selected_tag=hybrid_relay_selected_tag,
|
||
sidecar_kind=sidecar_kind,
|
||
session_generation=self._session_generation,
|
||
runtime_kind=runtime_kind,
|
||
sidecar_process_generation=(
|
||
int(getattr(self, "_hysteria_active_generation", 0))
|
||
if sidecar_kind == "hysteria"
|
||
else 0
|
||
),
|
||
relay_host=PROXY_HOST,
|
||
relay_credentials_generation=int(getattr(self, "_relay_credentials_generation", 0)),
|
||
front_process_generation=int(getattr(self, "_front_process_generation", 0)),
|
||
front_target_generation=int(getattr(self, "_front_target_generation", 0)),
|
||
started_at_monotonic=started_at,
|
||
ready_at_monotonic=ready_at,
|
||
failure_episode_id=int(getattr(self, "_hysteria_failure_episode_id", 0)),
|
||
last_failure_code=getattr(self, "_hysteria_last_failure_code", None),
|
||
automatic_switch_attempted=bool(
|
||
getattr(self, "_hysteria_automatic_switch_attempted", False)
|
||
),
|
||
)
|
||
if isinstance(contract, HysteriaTransitionContract):
|
||
if contract.session.session_generation != self._session_generation:
|
||
contract.begin(
|
||
self._session_generation,
|
||
node.id if node else None,
|
||
runtime_kind,
|
||
preserve_failure_episode=bool(
|
||
getattr(self, "_hysteria_recovery_active", False)
|
||
),
|
||
)
|
||
contract.session.ready_at_monotonic = ready_at
|
||
contract.session.state = HysteriaRuntimeState.READY
|
||
if hasattr(self, "_hysteria_recovery_active"):
|
||
self._hysteria_recovery_active = False
|
||
if hasattr(self, "_hysteria_automatic_switch_attempted"):
|
||
self._hysteria_automatic_switch_attempted = False
|
||
self._blocked_transition_signature = ""
|
||
begin_auto_switch_warmup(self, node)
|
||
|
||
def _derive_outbound_pool_tags(self, node: Node | None, *, core: str) -> dict[str, str] | None:
|
||
"""П1 (AC1/A2): вывести дефолтные outbound_pool_tags из пула контроллера.
|
||
|
||
Теги честны относительно того, что реально загрузило запущенное ядро:
|
||
xray/tun2socks встраивают пул в конфиг ровно тогда, когда выбранная нода
|
||
входит в пул (условие сборщиков конфига). Если пул не загружался (нет
|
||
ноды, нода вне пула) — тегов нет. Для sing-box действует другая схема
|
||
тегов (selector_tags): их вызывающие обязаны передавать явно, выводить
|
||
их из Xray-пула нельзя.
|
||
"""
|
||
|
||
if node is None or core not in {"xray", "tun2socks"}:
|
||
return None
|
||
pool = self.xray_outbound_pool()
|
||
if not pool.contains(node.id):
|
||
return None
|
||
return dict(pool.tags)
|
||
|
||
def _clear_active_session(self) -> None:
|
||
self._active_session = None
|
||
self._core_log_contexts.clear()
|
||
|
||
# ── System proxy: WinAPI only in the FIFO proxy thread ──
|
||
|
||
def _proxy_steps(self, method: str, *args: Any, **kwargs: Any) -> TransitionSteps:
|
||
"""Await one ProxyManager call in the ordered proxy thread."""
|
||
call = getattr(self.proxy, method)
|
||
try:
|
||
return (yield run_in_worker(lambda: call(*args, **kwargs), executor=PROXY_EXECUTOR))
|
||
finally:
|
||
self._emit_system_proxy_state()
|
||
|
||
def _proxy_nowait(self, method: str, *args: Any, **kwargs: Any) -> None:
|
||
"""Queue one ProxyManager call without waiting (signal handlers)."""
|
||
call = getattr(self.proxy, method)
|
||
future = PROXY_EXECUTOR.submit(call, *args, **kwargs)
|
||
|
||
def _report(done) -> None:
|
||
error = done.exception()
|
||
try:
|
||
# Поток очереди прокси → GUI-поток (queued): там перечитаем реестр.
|
||
self._proxy_operation_done.emit()
|
||
if error is not None:
|
||
self._background_log.emit(f"[proxy] {method} failed: {error!r}")
|
||
except RuntimeError:
|
||
pass
|
||
|
||
future.add_done_callback(_report)
|
||
|
||
def _proxy_blocking(self, method: str, *args: Any, timeout: float = 10.0, **kwargs: Any) -> Any:
|
||
"""Shutdown only: run after everything already queued, and wait."""
|
||
future = PROXY_EXECUTOR.submit(getattr(self.proxy, method), *args, **kwargs)
|
||
try:
|
||
return future.result(timeout=timeout)
|
||
except Exception as exc: # noqa: BLE001 — shutdown must continue
|
||
self._log(f"[proxy] {method} failed during shutdown: {exc!r}")
|
||
return None
|
||
|
||
def _apply_proxy_runtime_change(self) -> bool:
|
||
"""Синхронный драйвер (тесты). Переходы — ``_apply_proxy_runtime_change_steps``."""
|
||
return bool(run_steps_blocking(self._apply_proxy_runtime_change_steps()))
|
||
|
||
def _apply_proxy_runtime_change_steps(self) -> TransitionSteps:
|
||
settings = self.state.settings
|
||
bypass_lan = self._system_proxy_bypass_lan()
|
||
if self._active_session is not None:
|
||
socks_port = self._active_session.socks_port
|
||
http_port = self._active_session.http_port
|
||
else:
|
||
socks_port, http_port = self.get_effective_proxy_ports()
|
||
try:
|
||
if settings.enable_system_proxy:
|
||
yield from self._proxy_steps(
|
||
"enable",
|
||
http_port,
|
||
socks_port,
|
||
bypass_lan=bypass_lan,
|
||
)
|
||
else:
|
||
yield from self._proxy_steps("disable", restore_previous=True)
|
||
except Exception as exc:
|
||
self._set_connection_status(
|
||
"error",
|
||
f"Не удалось применить системный прокси: {exc}",
|
||
level="error",
|
||
)
|
||
return False
|
||
|
||
node = self.selected_node
|
||
if self.connected:
|
||
session = self._active_session
|
||
self._capture_active_session(
|
||
node,
|
||
tun=False,
|
||
core=session.active_core if session is not None else self._active_core,
|
||
api_port=session.api_port if session is not None else self._xray_api_port,
|
||
hybrid=session.hybrid if session is not None else False,
|
||
sidecar_kind=session.sidecar_kind if session is not None else "",
|
||
socks_port=socks_port,
|
||
http_port=http_port,
|
||
xray_inbound_tags=session.xray_inbound_tags if session is not None else (),
|
||
sidecar_relay_port=session.sidecar_relay_port if session is not None else 0,
|
||
protect_ss_port=session.protect_ss_port if session is not None else 0,
|
||
protect_ss_password=session.protect_ss_password if session is not None else "",
|
||
ping_host=session.ping_host if session is not None else "",
|
||
ping_port=session.ping_port if session is not None else 0,
|
||
# A2/AC4: без снапшота сессии нельзя утверждать, что ядро грузило
|
||
# пул — пустое переопределение честнее выведенного дефолта.
|
||
outbound_pool_tags=session.outbound_pool_tags if session is not None else {},
|
||
hybrid_relay_selector_tags=(
|
||
session.hybrid_relay_selector_tags if session is not None else ()
|
||
),
|
||
hybrid_relay_selected_tag=(
|
||
session.hybrid_relay_selected_tag if session is not None else ""
|
||
),
|
||
)
|
||
return True
|
||
|
||
def _needs_transition(self) -> bool:
|
||
node = self._runtime_selected_node()
|
||
context = TransitionContext(
|
||
desired_connected=self._desired_connected,
|
||
locked=self.locked,
|
||
has_selected_node=node is not None,
|
||
can_connect_without_selected_node=self._can_connect_without_selected_node(),
|
||
connected=self.connected,
|
||
blocked_transition_signature=self._blocked_transition_signature,
|
||
current_transition_signature=self._transition_signature(node),
|
||
active_session=self._active_session,
|
||
can_apply_proxy_runtime_change=False,
|
||
can_tun_hot_swap=False,
|
||
can_proxy_hot_swap=False,
|
||
)
|
||
return needs_transition(context)
|
||
|
||
def _can_apply_proxy_runtime_change(self, session: ActiveSessionSnapshot) -> bool:
|
||
settings = self.state.settings
|
||
desired_core = "singbox" if self.is_singbox_proxy_mode(settings) else "xray"
|
||
if session.active_core != desired_core:
|
||
return False
|
||
return can_apply_proxy_runtime_change_rule(
|
||
session=session,
|
||
settings_tun_mode=bool(settings.tun_mode),
|
||
current_xray_layer_signature=self._xray_layer_signature(),
|
||
proxy_enabled=bool(settings.enable_system_proxy),
|
||
proxy_bypass_lan=self._system_proxy_bypass_lan(),
|
||
)
|
||
|
||
def _can_proxy_hot_swap(self, session: ActiveSessionSnapshot) -> bool:
|
||
settings = self.state.settings
|
||
desired_core = "singbox" if self.is_singbox_proxy_mode(settings) else "xray"
|
||
if session.active_core != desired_core:
|
||
return False
|
||
if desired_core == "singbox":
|
||
socks_port, http_port = session.socks_port, session.http_port
|
||
else:
|
||
_, _, _, socks_port, http_port, _ = self._inspect_active_xray_config()
|
||
return can_proxy_hot_swap_rule(
|
||
session=session,
|
||
settings_tun_mode=bool(settings.tun_mode),
|
||
socks_port=int(socks_port),
|
||
http_port=int(http_port),
|
||
current_xray_layer_signature=self._xray_layer_signature(),
|
||
)
|
||
|
||
def _can_tun_hot_swap(self, session: ActiveSessionSnapshot) -> bool:
|
||
settings = self.state.settings
|
||
node = self._runtime_selected_node()
|
||
return can_tun_hot_swap_rule(
|
||
session=session,
|
||
settings_tun_mode=bool(settings.tun_mode),
|
||
settings_tun_engine=str(settings.tun_engine),
|
||
has_selected_node=node is not None,
|
||
current_tun_layer_signature=self._tun_layer_signature(node, settings, self.state.routing),
|
||
)
|
||
|
||
def _compute_transition_action(self) -> str | None:
|
||
node = self._runtime_selected_node()
|
||
session = self._active_session
|
||
context = TransitionContext(
|
||
desired_connected=self._desired_connected,
|
||
locked=self.locked,
|
||
has_selected_node=node is not None,
|
||
can_connect_without_selected_node=self._can_connect_without_selected_node(),
|
||
connected=self.connected,
|
||
blocked_transition_signature=self._blocked_transition_signature,
|
||
current_transition_signature=self._transition_signature(node),
|
||
active_session=session,
|
||
can_apply_proxy_runtime_change=self._can_apply_proxy_runtime_change(session) if session is not None else False,
|
||
can_tun_hot_swap=self._can_tun_hot_swap(session) if session is not None else False,
|
||
can_proxy_hot_swap=self._can_proxy_hot_swap(session) if session is not None else False,
|
||
)
|
||
return compute_transition_action(context)
|
||
|
||
def _transition_status_text(self, action: str) -> str:
|
||
return transition_status_text(action)
|
||
|
||
def _request_transition(self, reason: str, *, keep_blocked: bool = False) -> None:
|
||
"""Единственный вход в координатор переходов.
|
||
|
||
Запрос только фиксирует желаемое состояние и новое поколение. Если
|
||
переход уже идёт, он сам увидит устаревшее поколение на ближайшем
|
||
шаге и завершится; подготовка (DNS/Zapret) и новый переход начнутся
|
||
после него (``_kick_transition``). Никаких вложенных остановок из
|
||
обработчиков кликов: все клики сливаются в последнее состояние.
|
||
"""
|
||
if not keep_blocked:
|
||
self._blocked_transition_signature = ""
|
||
if getattr(self, "_transition_requested_at", None) is None:
|
||
# Часы первого из слитых запросов: «нажал → готово» в [perf].
|
||
self._transition_requested_at = time.perf_counter()
|
||
self._transition_pending = True
|
||
self._transition_reason = reason
|
||
self._transition_generation += 1
|
||
self.bypass.cancel_wait()
|
||
if self._transition_active:
|
||
return
|
||
self._kick_transition(transition_request_delay_ms(reason))
|
||
|
||
def _kick_transition(self, delay_ms: int = 0) -> None:
|
||
"""Run the selected-server preparation once per generation, then drain."""
|
||
generation = self._transition_generation
|
||
if self._desired_connected and self._protection_prepared_generation != generation:
|
||
self._protection_prepared_generation = generation
|
||
if self.bypass.prepare(generation):
|
||
if self._transition_timer.isActive():
|
||
self._transition_timer.stop()
|
||
self._transition_scheduled = False
|
||
return
|
||
self._schedule_transition_drain(delay_ms)
|
||
|
||
def _run_prestop(
|
||
self,
|
||
generation: int,
|
||
on_stopped: Callable[[], None],
|
||
*,
|
||
failure_message: str | None,
|
||
disable_proxy: bool = False,
|
||
) -> None:
|
||
"""Stop the data plane as a coordinator-owned step transition.
|
||
|
||
Used where the selected-server preparation must stop the old tunnel
|
||
before DNS/winws2 changes. The stop runs through TransitionRunner
|
||
(``_transition_active`` holds the queue), never synchronously inside a
|
||
click or signal handler.
|
||
"""
|
||
self._transition_active = True
|
||
self.bypass.hold(generation)
|
||
runner = TransitionRunner(
|
||
self._disconnect_current_steps(disable_proxy=disable_proxy, emit_status=False),
|
||
on_finished=lambda finished: self._on_prestop_finished(
|
||
finished, generation, on_stopped, failure_message
|
||
),
|
||
parent=self,
|
||
)
|
||
self._transition_runner = runner
|
||
runner.start()
|
||
|
||
def _on_prestop_finished(
|
||
self,
|
||
runner: TransitionRunner,
|
||
generation: int,
|
||
on_stopped: Callable[[], None],
|
||
failure_message: str | None,
|
||
) -> None:
|
||
if self._transition_runner is runner:
|
||
self._transition_runner = None
|
||
runner.deleteLater()
|
||
_log_transition_profile(self, "pre-stop", runner)
|
||
self._transition_active = False
|
||
if runner.cancelled:
|
||
return
|
||
if runner.error is not None:
|
||
self._log(f"[transition] pre-stop failed with error: {runner.error!r}")
|
||
stopped = bool(runner.result) and runner.error is None
|
||
if generation != self._transition_generation:
|
||
# A newer request arrived while stopping: prepare for it instead.
|
||
self._kick_transition(0)
|
||
return
|
||
self.bypass.cancel_wait()
|
||
if not stopped and failure_message is not None:
|
||
self._cancel_target_transition(failure_message)
|
||
return
|
||
self._desired_connected = True
|
||
on_stopped()
|
||
|
||
def _run_coordinated_stop(self, reason: str, on_stopped: Callable[[bool], None]) -> bool:
|
||
"""Disconnect as a coordinator-owned step transition, then call back.
|
||
|
||
For actions that need the connection fully stopped before they run
|
||
(Xray core replacement). ``False`` — a transition is in flight; the
|
||
caller asks the user to retry instead of nesting a stop into it.
|
||
"""
|
||
if self._transition_active:
|
||
return False
|
||
self._log(f"[transition] coordinated stop: {reason}")
|
||
self._desired_connected = False
|
||
self._transition_generation += 1
|
||
self._transition_pending = False
|
||
if self._transition_timer.isActive():
|
||
self._transition_timer.stop()
|
||
self._transition_scheduled = False
|
||
self._transition_active = True
|
||
self.transition_state_changed.emit(True, self._transition_status_text("disconnect"))
|
||
runner = TransitionRunner(
|
||
self._disconnect_current_steps(),
|
||
on_finished=lambda finished: self._on_coordinated_stop_finished(finished, on_stopped),
|
||
parent=self,
|
||
)
|
||
self._transition_runner = runner
|
||
runner.start()
|
||
return True
|
||
|
||
def _on_coordinated_stop_finished(
|
||
self,
|
||
runner: TransitionRunner,
|
||
on_stopped: Callable[[bool], None],
|
||
) -> None:
|
||
if self._transition_runner is runner:
|
||
self._transition_runner = None
|
||
runner.deleteLater()
|
||
self._transition_active = False
|
||
if runner.cancelled:
|
||
return
|
||
if runner.error is not None:
|
||
self._log(f"[transition] coordinated stop failed with error: {runner.error!r}")
|
||
try:
|
||
on_stopped(bool(runner.result) and runner.error is None)
|
||
finally:
|
||
if self._transition_pending or self._cleanup_pending or self._needs_transition():
|
||
self._kick_transition(0)
|
||
else:
|
||
self.transition_state_changed.emit(False, "")
|
||
|
||
def _request_stop(self, reason: str, *, quiet: bool) -> None:
|
||
"""Queue a coordinated disconnect (desired state is already False)."""
|
||
self._desired_connected = False
|
||
if quiet:
|
||
self._quiet_disconnect = True
|
||
self._request_transition(reason, keep_blocked=True)
|
||
|
||
def _active_config_uses_selected_node(self, node: Node | None) -> bool:
|
||
"""Avoid targeting a node ignored by the active raw JSON document."""
|
||
if node is None:
|
||
return False
|
||
try:
|
||
if self.is_singbox_editor_mode():
|
||
facts = self._singbox_plan_facts(node, tun=bool(self.state.settings.tun_mode))
|
||
if facts is None:
|
||
# The normal planner will show its specific error later.
|
||
return True
|
||
return facts.used_selected_node
|
||
if self.uses_xray_raw_config():
|
||
runtime = self._build_runtime_xray_config(
|
||
node,
|
||
tun_mode=bool(self.state.settings.tun_mode),
|
||
)
|
||
return bool(runtime.used_selected_node)
|
||
except (OSError, ValueError):
|
||
# The normal planner will show its specific error later. Do not
|
||
# accidentally weaken the gate because preflight itself failed.
|
||
return True
|
||
return True
|
||
|
||
def _cancel_target_transition(self, message: str) -> None:
|
||
self.bypass.cancel_wait()
|
||
self._transition_pending = False
|
||
if self._hysteria_recovery_active:
|
||
self._clear_pending_transport_selection()
|
||
self._hysteria_recovery_active = False
|
||
self._desired_connected = False
|
||
self._handle_unexpected_disconnect()
|
||
else:
|
||
self._desired_connected = self.connected
|
||
self._blocked_transition_signature = self._transition_signature()
|
||
self._set_connection_status("error", message, level="warning")
|
||
self.transition_state_changed.emit(False, "")
|
||
|
||
def target_profile_allows_core_start(
|
||
self,
|
||
node: Node | None,
|
||
*,
|
||
used_selected_node: bool = True,
|
||
) -> bool:
|
||
"""Last-moment fence immediately before any VPN process start."""
|
||
return self.bypass.allows_core_start(node, used_selected_node=used_selected_node)
|
||
|
||
def start_zapret(self, preset_name: str) -> None:
|
||
"""Start winws2 on a preset with a freshly resolved selected-server rule."""
|
||
self.bypass.start_manual(preset_name)
|
||
|
||
def stop_zapret(self) -> None:
|
||
"""Stop winws2; a pending manual start is cancelled with it."""
|
||
self.bypass.stop_manual()
|
||
|
||
def apply_zapret_target_settings(self, settings: ZapretTargetSettings) -> None:
|
||
"""New per-kind bypass choices: bring the connection or winws2 in line."""
|
||
self.state.settings.zapret_target = settings
|
||
self.schedule_save()
|
||
if self.connected or self._desired_connected:
|
||
self._desired_connected = True
|
||
self._request_transition("Zapret target strategy changed")
|
||
elif self.zapret.plan is not None:
|
||
self.start_zapret(self.state.settings.zapret_preset or self.zapret.plan.preset)
|
||
|
||
def _schedule_transition_drain(self, delay_ms: int) -> None:
|
||
if self._transition_active or self.bypass.waiting_for(self._transition_generation):
|
||
return
|
||
self._transition_scheduled = True
|
||
self._transition_timer.start(max(0, int(delay_ms)))
|
||
|
||
def _drain_transition_queue(self) -> None:
|
||
self._transition_scheduled = False
|
||
if self._transition_active:
|
||
return
|
||
if self.bypass.waiting_for(self._transition_generation):
|
||
return
|
||
|
||
if self._cleanup_pending or self._has_residual_processes():
|
||
# Процессы недостартовавшего/прерванного перехода гасятся отдельным
|
||
# шагом до любого нового действия; следующий drain решит, что дальше.
|
||
self._cleanup_pending = False
|
||
action, reason = "cleanup", "cleanup"
|
||
else:
|
||
if not self._transition_pending and not self._needs_transition():
|
||
self.transition_state_changed.emit(False, "")
|
||
return
|
||
|
||
action = self._compute_transition_action()
|
||
if action is None:
|
||
self._transition_pending = False
|
||
self.transition_state_changed.emit(False, "")
|
||
return
|
||
|
||
self._transition_pending = False
|
||
reason = self._transition_reason or action
|
||
self._transition_active = True
|
||
self.transition_state_changed.emit(True, self._transition_status_text(action))
|
||
# Переход выполняется генератором через TransitionRunner: ожидания
|
||
# процессов — сигналы QProcess, блокирующий I/O — воркеры, паузы —
|
||
# таймеры; GUI-поток между шагами свободен. Устаревание поколения
|
||
# генераторы проверяют сами после каждого шага и сами прибирают
|
||
# недостартовавшее (явные ветки отмены), поэтому раннер их не закрывает.
|
||
generation = self._transition_generation
|
||
runner = TransitionRunner(
|
||
self._transition_action_steps(action, reason),
|
||
on_finished=lambda finished, generation=generation, action=action: (
|
||
self._on_transition_runner_finished(finished, generation, action)
|
||
),
|
||
parent=self,
|
||
)
|
||
runner.requested_at = getattr(self, "_transition_requested_at", None)
|
||
self._transition_requested_at = None
|
||
self._transition_runner = runner
|
||
runner.start()
|
||
|
||
def _on_transition_runner_finished(
|
||
self,
|
||
runner: TransitionRunner,
|
||
generation: int | None = None,
|
||
action: str = "",
|
||
) -> None:
|
||
if self._transition_runner is runner:
|
||
self._transition_runner = None
|
||
runner.deleteLater()
|
||
_log_transition_profile(self, f"transition {action}", runner)
|
||
superseded = runner.cancelled or (
|
||
generation is not None and generation != self._transition_generation
|
||
)
|
||
try:
|
||
if runner.error is not None:
|
||
self._log(f"[transition] failed with error: {runner.error!r}")
|
||
if action != "cleanup" and not superseded:
|
||
ok = bool(runner.result) and runner.error is None
|
||
if ok:
|
||
self._blocked_transition_signature = ""
|
||
else:
|
||
self._blocked_transition_signature = self._transition_signature(
|
||
self._runtime_selected_node()
|
||
)
|
||
self._desired_connected = self.connected
|
||
elif superseded and not runner.cancelled:
|
||
# Устаревший переход не блокирует сигнатуру и не трогает
|
||
# desired_connected (их задаёт последний запрос), но обязан
|
||
# оставить связку процессов консистентной.
|
||
self._reconcile_cancelled_transition()
|
||
if (
|
||
not runner.cancelled
|
||
and not self.connected
|
||
and not getattr(self, "_hysteria_recovery_active", False)
|
||
and self._has_residual_processes()
|
||
):
|
||
self._cleanup_pending = True
|
||
finally:
|
||
self._transition_active = False
|
||
if not superseded and self._pending_transport_node_id:
|
||
self._clear_pending_transport_selection()
|
||
self._hysteria_recovery_active = False
|
||
if runner.cancelled:
|
||
pass
|
||
elif self._transition_pending or self._cleanup_pending or self._needs_transition():
|
||
self._kick_transition(0)
|
||
else:
|
||
self.transition_state_changed.emit(False, "")
|
||
|
||
def _has_residual_processes(self) -> bool:
|
||
"""Processes alive without a connected session (aborted transition)."""
|
||
if self.connected:
|
||
return False
|
||
for manager in (self.singbox, self.xray, self.hysteria, getattr(self, "amnezia", None)):
|
||
if manager is None:
|
||
continue
|
||
if manager.is_running is True or _process_alive(manager):
|
||
return True
|
||
return False
|
||
|
||
def _reconcile_cancelled_transition(self) -> None:
|
||
"""Привести состояние к консистентному виду после устаревшего перехода.
|
||
|
||
Устаревший переход мог остановиться между шагами: например транспорт
|
||
уже остановлен, а sing-box ещё жив (connected=False). Сессия без
|
||
живого соединения очищается сразу; остановку оставшихся процессов
|
||
выполнит отдельный шаг ``cleanup`` в очереди (не синхронно здесь).
|
||
"""
|
||
if self.connected:
|
||
return
|
||
if self._has_residual_processes():
|
||
self._log("[transition] superseded mid-swap — partial connection processes will be stopped")
|
||
self._cleanup_pending = True
|
||
if self._active_session is not None:
|
||
self._clear_active_session()
|
||
if not self._desired_connected:
|
||
self._set_connection_status("idle", "Отключено", level="info")
|
||
|
||
def _cleanup_residual_steps(self) -> TransitionSteps:
|
||
self._log("[transition] stopping partial connection processes")
|
||
stopped = yield from self._stop_active_connection_processes_steps(
|
||
disable_proxy=not self._desired_connected
|
||
)
|
||
self._refresh_connected_state()
|
||
if not self.connected and self._active_session is not None:
|
||
self._clear_active_session()
|
||
if not self._desired_connected:
|
||
self._set_connection_status("idle", "Отключено", level="info")
|
||
return stopped
|
||
|
||
def _transition_action_steps(self, action: str, reason: str) -> TransitionSteps:
|
||
"""Generator executed by TransitionRunner for one transition action.
|
||
|
||
Все действия генераторные: ожидания процессов, control-plane, пробы,
|
||
системный прокси и паузы — шаги; GUI-поток не блокируется и не
|
||
прокачивает события вложенно.
|
||
"""
|
||
if action == "cleanup":
|
||
return (yield from self._cleanup_residual_steps())
|
||
if action == "disconnect":
|
||
quiet = bool(getattr(self, "_quiet_disconnect", False))
|
||
self._quiet_disconnect = False
|
||
return (yield from self._disconnect_current_steps(disable_proxy=True, emit_status=not quiet))
|
||
if action == "connect":
|
||
return (yield from self._connect_selected_steps())
|
||
if action == "proxy_update":
|
||
return (yield from self._apply_proxy_runtime_change_steps())
|
||
if action == "tun_hot_swap":
|
||
return (yield from self._hot_swap_node_steps(reason))
|
||
if action == "proxy_hot_swap":
|
||
return (yield from self._restart_proxy_core_steps(reason))
|
||
return (yield from self._reconnect_steps(reason))
|
||
|
||
def _run_transition_action(self, action: str, reason: str) -> bool:
|
||
"""Синхронный драйвер действия (тесты)."""
|
||
return bool(run_steps_blocking(self._transition_action_steps(action, reason)))
|
||
|
||
# ── Country detection helpers ──
|
||
|
||
def _detect_countries_sync(self) -> None:
|
||
detect_countries_sync_operation(self)
|
||
|
||
def _start_country_ip_resolution(self) -> None:
|
||
start_country_ip_resolution_operation(self)
|
||
|
||
def _on_countries_resolved(self, results: dict[str, str]) -> None:
|
||
on_countries_resolved_operation(self, results)
|
||
|
||
def shutdown(self) -> None:
|
||
# Незавершённый асинхронный переход закрывается (finally-блоки операций
|
||
# выполняются) до остановки процессов при выходе.
|
||
self._cancel_hot_switch_runner()
|
||
runner = self._transition_runner
|
||
if runner is not None:
|
||
runner.cancel()
|
||
shutdown_operation(self)
|
||
|
||
@staticmethod
|
||
def _cleanup_tun_adapter() -> None:
|
||
"""Remove the wintun TUN adapter if it was left behind."""
|
||
import subprocess as _sp
|
||
try:
|
||
result = run_text(
|
||
["netsh", "interface", "show", "interface"],
|
||
timeout=5,
|
||
creationflags=0x08000000,
|
||
)
|
||
if "ZapretKVN_TUN" in result_output_text(result):
|
||
_sp.run(
|
||
["netsh", "interface", "set", "interface", "ZapretKVN_TUN", "admin=disable"],
|
||
capture_output=True, timeout=5,
|
||
creationflags=0x08000000,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
@property
|
||
def selected_node(self) -> Node | None:
|
||
return self._get_node_by_id(self.state.selected_node_id)
|
||
|
||
def _runtime_selected_node(self) -> Node | None:
|
||
pending = getattr(self, "_pending_transport_node_id", None)
|
||
if isinstance(pending, str) and pending:
|
||
return self._get_node_by_id(pending)
|
||
return self.selected_node
|
||
|
||
def _commit_pending_transport_selection(self, node: Node | None) -> bool:
|
||
"""Commit a ready runtime's selection through one path for all transports."""
|
||
pending = getattr(self, "_pending_transport_node_id", None)
|
||
if not isinstance(pending, str) or not pending:
|
||
return True
|
||
if node is None or node.id != pending:
|
||
self._log("[transport-transition] ready runtime does not match pending selection")
|
||
return False
|
||
self.state.selected_node_id = node.id
|
||
self._clear_pending_transport_selection()
|
||
self.selection_changed.emit(node)
|
||
return True
|
||
|
||
def _clear_pending_transport_selection(self) -> None:
|
||
self._pending_transport_node_id = None
|
||
|
||
def _get_node_by_id(self, node_id: str | None) -> Node | None:
|
||
return get_node_by_id_operation(self, node_id)
|
||
|
||
def _prepare_node_for_runtime(self, node: Node | None) -> str | None:
|
||
return prepare_node_for_runtime_operation(self, node)
|
||
|
||
def export_node_outbound_json(self, node_id: str | None = None) -> str | None:
|
||
node = self._get_node_by_id(node_id) if node_id else self.selected_node
|
||
if not node:
|
||
return None
|
||
return json.dumps(node.outbound, ensure_ascii=True, indent=2)
|
||
|
||
def export_runtime_config_json(self, node_id: str | None = None) -> str | None:
|
||
node = self._get_node_by_id(node_id) if node_id else self.selected_node
|
||
try:
|
||
if self.is_singbox_editor_mode():
|
||
plan = (
|
||
self._plan_runtime_singbox(node)
|
||
if self.is_singbox_tun_mode()
|
||
else self._plan_proxy_runtime_singbox(node)
|
||
)
|
||
return json.dumps(plan.singbox_config, ensure_ascii=True, indent=2)
|
||
if self.uses_xray_raw_config():
|
||
runtime = self._build_runtime_xray_config(node, tun_mode=self.is_xray_tun_mode())
|
||
return json.dumps(runtime.config, ensure_ascii=True, indent=2)
|
||
if not node:
|
||
return None
|
||
problem = self._prepare_node_for_runtime(node)
|
||
if problem:
|
||
return None
|
||
cfg = build_xray_config(
|
||
node,
|
||
self.state.routing,
|
||
self.state.settings,
|
||
socks_port=DEFAULT_SOCKS_PORT,
|
||
http_port=DEFAULT_HTTP_PORT,
|
||
)
|
||
return json.dumps(cfg, ensure_ascii=True, indent=2)
|
||
except ValueError:
|
||
return None
|
||
|
||
def import_nodes_from_text(self, text: str) -> tuple[int, list[str]]:
|
||
return import_nodes_from_text_operation(self, text)
|
||
|
||
def get_subscription(self, subscription_id: str | None) -> Subscription | None:
|
||
if not subscription_id:
|
||
return None
|
||
return next((item for item in self.state.subscriptions if item.id == subscription_id), None)
|
||
|
||
def _unique_subscription_name(self, name: str, *, exclude_id: str | None = None) -> str:
|
||
base = name.strip() or "Подписка"
|
||
used = {
|
||
item.name.casefold()
|
||
for item in self.state.subscriptions
|
||
if item.id != exclude_id and item.name.strip()
|
||
}
|
||
if base.casefold() not in used:
|
||
return base
|
||
index = 2
|
||
while f"{base} ({index})".casefold() in used:
|
||
index += 1
|
||
return f"{base} ({index})"
|
||
|
||
def add_subscription(self, subscription: Subscription, *, mode: str = "auto") -> bool:
|
||
subscription.url, profile_hint = resolve_subscription_source(subscription.url)
|
||
if profile_hint and normalize_client_profile(subscription.client_profile) == "zapret":
|
||
subscription.client_profile = profile_hint
|
||
self._normalize_subscription_identity(subscription)
|
||
validate_filter_patterns(subscription.include_pattern, subscription.exclude_pattern)
|
||
if any(item.url == subscription.url for item in self.state.subscriptions):
|
||
self.status.emit("warning", "Эта подписка уже добавлена")
|
||
return False
|
||
if subscription.id in self._subscription_workers or subscription.id in self._subscription_queued_ids:
|
||
return False
|
||
self._pending_subscription_additions[subscription.id] = subscription
|
||
self._enqueue_subscription_update(subscription, mode, pending_add=True, apply_result=True)
|
||
return True
|
||
|
||
def update_subscription_definition(self, subscription_id: str, updates: dict[str, Any]) -> bool:
|
||
subscription = self.get_subscription(subscription_id)
|
||
if (
|
||
subscription is None
|
||
or subscription_id in self._subscription_workers
|
||
or subscription_id in self._subscription_queued_ids
|
||
):
|
||
return False
|
||
url = str(updates.get("url", subscription.url)).strip()
|
||
include_pattern = str(updates.get("include_pattern", subscription.include_pattern)).strip()
|
||
exclude_pattern = str(updates.get("exclude_pattern", subscription.exclude_pattern)).strip()
|
||
url, profile_hint = resolve_subscription_source(url)
|
||
validate_filter_patterns(include_pattern, exclude_pattern)
|
||
if any(item.id != subscription_id and item.url == url for item in self.state.subscriptions):
|
||
self.status.emit("warning", "Эта подписка уже добавлена")
|
||
return False
|
||
old_url = subscription.url
|
||
old_include_pattern = subscription.include_pattern
|
||
old_exclude_pattern = subscription.exclude_pattern
|
||
old_identity = (
|
||
subscription.client_profile,
|
||
subscription.user_agent,
|
||
subscription.send_hwid,
|
||
subscription.hwid,
|
||
)
|
||
requested_profile = str(updates.get("client_profile", subscription.client_profile))
|
||
normalized_profile = normalize_client_profile(
|
||
profile_hint
|
||
if profile_hint and normalize_client_profile(requested_profile) == "zapret"
|
||
else requested_profile
|
||
)
|
||
send_hwid = bool(updates.get("send_hwid", subscription.send_hwid))
|
||
hwid = str(updates.get("hwid", subscription.hwid)).strip()
|
||
if send_hwid:
|
||
hwid = validate_hwid(hwid or self.state.subscription_device_id)
|
||
interval = updates.get("update_interval_hours", subscription.update_interval_hours)
|
||
normalized_interval = int(interval) if interval else None
|
||
|
||
requested_name = str(updates.get("name", subscription.name)).strip()
|
||
subscription.name = self._unique_subscription_name(requested_name, exclude_id=subscription_id)
|
||
subscription.url = url
|
||
subscription.user_agent = str(updates.get("user_agent", subscription.user_agent)).strip()
|
||
subscription.client_profile = normalized_profile
|
||
subscription.send_hwid = send_hwid
|
||
subscription.hwid = hwid
|
||
subscription.auto_update = bool(updates.get("auto_update", subscription.auto_update))
|
||
subscription.update_interval_hours = normalized_interval
|
||
subscription.include_pattern = include_pattern
|
||
subscription.exclude_pattern = exclude_pattern
|
||
if (
|
||
old_url != url
|
||
or old_include_pattern != include_pattern
|
||
or old_exclude_pattern != exclude_pattern
|
||
or old_identity
|
||
!= (
|
||
subscription.client_profile,
|
||
subscription.user_agent,
|
||
subscription.send_hwid,
|
||
subscription.hwid,
|
||
)
|
||
):
|
||
subscription.etag = ""
|
||
subscription.last_modified = ""
|
||
subscription.pending_url = ""
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
return True
|
||
|
||
def _normalize_subscription_identity(self, subscription: Subscription) -> None:
|
||
subscription.client_profile = normalize_client_profile(subscription.client_profile)
|
||
if not subscription.send_hwid:
|
||
subscription.hwid = subscription.hwid.strip()
|
||
return
|
||
subscription.hwid = validate_hwid(
|
||
subscription.hwid or self.state.subscription_device_id
|
||
)
|
||
|
||
def update_subscription(
|
||
self,
|
||
subscription_id: str,
|
||
*,
|
||
mode: str = "auto",
|
||
force_refresh: bool = False,
|
||
) -> bool:
|
||
subscription = self.get_subscription(subscription_id)
|
||
if subscription is None:
|
||
return False
|
||
return self._enqueue_subscription_update(
|
||
subscription,
|
||
mode,
|
||
pending_add=False,
|
||
apply_result=True,
|
||
force_refresh=force_refresh,
|
||
)
|
||
|
||
def check_subscription(self, subscription_id: str, *, mode: str = "auto") -> bool:
|
||
subscription = self.get_subscription(subscription_id)
|
||
if subscription is None:
|
||
return False
|
||
return self._enqueue_subscription_update(subscription, mode, pending_add=False, apply_result=False)
|
||
|
||
def update_all_subscriptions(self, *, mode: str = "auto") -> int:
|
||
count = 0
|
||
for subscription in sorted(self.state.subscriptions, key=lambda item: item.sort_order):
|
||
if self._enqueue_subscription_update(
|
||
subscription, mode, pending_add=False, apply_result=True
|
||
):
|
||
count += 1
|
||
return count
|
||
|
||
def remove_subscription(self, subscription_id: str, *, keep_nodes: bool = False) -> bool:
|
||
if subscription_id in self._subscription_workers:
|
||
self.status.emit("warning", "Дождитесь завершения обновления подписки")
|
||
return False
|
||
if subscription_id in self._subscription_queued_ids:
|
||
self._subscription_update_queue = [
|
||
item for item in self._subscription_update_queue if item[0].id != subscription_id
|
||
]
|
||
self._subscription_queued_ids.discard(subscription_id)
|
||
previous_selected = self.state.selected_node_id
|
||
selected_before = selection_snapshot(self)
|
||
if not remove_subscription_operation(self.state, subscription_id, keep_nodes=keep_nodes):
|
||
return False
|
||
log_selection_change(self, SelectionSource.SUBSCRIPTION, selected_before, self.state.selected_node_id)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.nodes_changed.emit(self.state.nodes)
|
||
self.selection_changed.emit(self.selected_node)
|
||
self.save()
|
||
if previous_selected != self.state.selected_node_id and (self.connected or self._desired_connected):
|
||
if self.state.selected_node_id is None and not self._can_connect_without_selected_node():
|
||
self._desired_connected = False
|
||
self._request_transition("subscription removed")
|
||
return True
|
||
|
||
def hide_subscription_node(self, node_id: str) -> bool:
|
||
return self.hide_subscription_nodes({node_id}) > 0
|
||
|
||
def hide_subscription_nodes(self, node_ids: set[str]) -> int:
|
||
previous_selected = self.state.selected_node_id
|
||
selected_before = selection_snapshot(self)
|
||
hidden = 0
|
||
for node_id in set(node_ids):
|
||
if hide_subscription_node_operation(self.state, node_id) is not None:
|
||
hidden += 1
|
||
if not hidden:
|
||
return 0
|
||
log_selection_change(self, SelectionSource.NODE_REMOVED, selected_before, self.state.selected_node_id)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.nodes_changed.emit(self.state.nodes)
|
||
self.selection_changed.emit(self.selected_node)
|
||
self.save()
|
||
if previous_selected != self.state.selected_node_id and (self.connected or self._desired_connected):
|
||
if self.state.selected_node_id is None and not self._can_connect_without_selected_node():
|
||
self._desired_connected = False
|
||
self._request_transition("subscription node hidden")
|
||
return hidden
|
||
|
||
def reset_subscription_hidden_nodes(self, subscription_id: str) -> bool:
|
||
subscription = self.get_subscription(subscription_id)
|
||
if (
|
||
subscription is None
|
||
or subscription_id in self._subscription_workers
|
||
or subscription_id in self._subscription_queued_ids
|
||
):
|
||
return False
|
||
subscription.hidden_source_keys.clear()
|
||
subscription.etag = ""
|
||
subscription.last_modified = ""
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
return True
|
||
|
||
def accept_subscription_pending_url(self, subscription_id: str) -> bool:
|
||
subscription = self.get_subscription(subscription_id)
|
||
if subscription is None or not subscription.pending_url:
|
||
return False
|
||
subscription.url = subscription.pending_url
|
||
subscription.pending_url = ""
|
||
subscription.etag = ""
|
||
subscription.last_modified = ""
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
return True
|
||
|
||
def _enqueue_subscription_update(
|
||
self,
|
||
subscription: Subscription,
|
||
mode: str,
|
||
pending_add: bool,
|
||
apply_result: bool,
|
||
force_refresh: bool = False,
|
||
) -> bool:
|
||
if subscription.id in self._subscription_workers or subscription.id in self._subscription_queued_ids:
|
||
return False
|
||
self._subscription_update_queue.append(
|
||
(deepcopy(subscription), mode, pending_add, apply_result, bool(force_refresh))
|
||
)
|
||
self._subscription_queued_ids.add(subscription.id)
|
||
self._drain_subscription_update_queue()
|
||
return True
|
||
|
||
def _drain_subscription_update_queue(self) -> None:
|
||
while self._subscription_update_queue and len(self._subscription_workers) < 3:
|
||
subscription, mode, pending_add, apply_result, force_refresh = (
|
||
self._subscription_update_queue.pop(0)
|
||
)
|
||
self._subscription_queued_ids.discard(subscription.id)
|
||
if not pending_add and self.get_subscription(subscription.id) is None:
|
||
continue
|
||
proxy_port = self.get_effective_http_proxy_port() if self.connected else None
|
||
worker = SubscriptionUpdateWorker(
|
||
subscription,
|
||
mode=mode,
|
||
proxy_port=proxy_port,
|
||
force_refresh=force_refresh,
|
||
parent=self,
|
||
)
|
||
self._subscription_workers[subscription.id] = worker
|
||
if not apply_result:
|
||
self._subscription_check_ids.add(subscription.id)
|
||
worker.progress.connect(self.subscription_update_progress.emit)
|
||
worker.completed.connect(self._on_subscription_update_completed)
|
||
worker.failed.connect(self._on_subscription_update_failed)
|
||
worker.finished.connect(
|
||
lambda sid=subscription.id, current=worker: self._on_subscription_worker_finished(sid, current)
|
||
)
|
||
self.subscription_update_started.emit(subscription.id)
|
||
worker.start()
|
||
|
||
def _on_subscription_update_completed(self, worker_subscription, fetched, parsed) -> None:
|
||
# Любой выход отсюда обязан завершиться subscription_update_finished,
|
||
# иначе строка подписки навсегда остаётся в «Обновление…».
|
||
try:
|
||
self._apply_subscription_update(worker_subscription, fetched, parsed)
|
||
except Exception as exc:
|
||
self._logger.exception("subscription update apply failed")
|
||
self._on_subscription_update_failed(
|
||
worker_subscription, f"Не удалось применить подписку: {exc}"
|
||
)
|
||
|
||
def _apply_subscription_update(self, worker_subscription, fetched, parsed) -> None:
|
||
self._log(
|
||
f"[subscription] {worker_subscription.name or 'новая'}: "
|
||
f"{'без изменений' if fetched.not_modified else 'загружена'} за {fetched.elapsed:.1f} с "
|
||
f"{'через прокси' if fetched.via_proxy else 'напрямую'}"
|
||
)
|
||
subscription = self.get_subscription(worker_subscription.id)
|
||
pending_add = subscription is None and worker_subscription.id in self._pending_subscription_additions
|
||
if pending_add:
|
||
if fetched.not_modified or parsed is None:
|
||
self._pending_subscription_additions.pop(worker_subscription.id, None)
|
||
self.subscription_update_finished.emit(
|
||
SubscriptionUpdateResult(
|
||
subscription_id=worker_subscription.id,
|
||
success=False,
|
||
message="Новая подписка вернула 304 без локального снимка",
|
||
)
|
||
)
|
||
return
|
||
subscription = self._pending_subscription_additions[worker_subscription.id]
|
||
subscription.name = self._unique_subscription_name(
|
||
subscription.name or parsed.metadata.title or "Подписка"
|
||
)
|
||
if subscription is None:
|
||
self.subscription_update_finished.emit(
|
||
SubscriptionUpdateResult(
|
||
subscription_id=worker_subscription.id,
|
||
success=False,
|
||
message="Подписка удалена во время обновления",
|
||
)
|
||
)
|
||
return
|
||
if parsed is not None and subscription.hidden_source_keys:
|
||
hidden = set(subscription.hidden_source_keys)
|
||
parsed.nodes = [node for node in parsed.nodes if node.source_key not in hidden]
|
||
if subscription.id in self._subscription_check_ids:
|
||
subscription.last_checked_at = datetime.now(timezone.utc).isoformat()
|
||
subscription.last_error = ""
|
||
subscription.failure_count = 0
|
||
subscription.backoff_until = None
|
||
node_count = len(
|
||
[node for node in self.state.nodes if node.subscription_id == subscription.id]
|
||
if fetched.not_modified
|
||
else parsed.nodes
|
||
)
|
||
stored_count = len(
|
||
[node for node in self.state.nodes if node.subscription_id == subscription.id]
|
||
)
|
||
result = SubscriptionUpdateResult(
|
||
subscription_id=subscription.id,
|
||
success=True,
|
||
message=f"Проверка успешна: в подписке серверов {node_count}",
|
||
skipped=(subscription.skipped_count if parsed is None else parsed.skipped),
|
||
warnings=(
|
||
list(subscription.warnings)
|
||
if parsed is None
|
||
else normalize_subscription_warnings(parsed.warnings)
|
||
),
|
||
not_modified=fetched.not_modified,
|
||
check_only=True,
|
||
source_count=node_count,
|
||
stored_count=stored_count,
|
||
)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
self.subscription_update_finished.emit(result)
|
||
return
|
||
if fetched.not_modified:
|
||
result = apply_not_modified(subscription, fetched)
|
||
else:
|
||
selected_before = selection_snapshot(self)
|
||
outcome = reconcile_subscription(self.state, subscription, parsed, fetched)
|
||
result = outcome.result
|
||
if pending_add:
|
||
self.state.subscriptions.append(subscription)
|
||
if pending_add and self.state.selected_node_id is None:
|
||
first = next((node for node in self.state.nodes if node.subscription_id == subscription.id), None)
|
||
self.state.selected_node_id = first.id if first else None
|
||
log_selection_change(self, SelectionSource.SUBSCRIPTION, selected_before, self.state.selected_node_id)
|
||
self._detect_countries_sync()
|
||
QTimer.singleShot(0, self._start_country_ip_resolution)
|
||
self.nodes_changed.emit(self.state.nodes)
|
||
self.selection_changed.emit(self.selected_node)
|
||
if result.reconnect_required and (self.connected or self._desired_connected):
|
||
if self.state.selected_node_id is None and not self._can_connect_without_selected_node():
|
||
self._desired_connected = False
|
||
else:
|
||
self._desired_connected = True
|
||
self._request_transition("subscription updated")
|
||
QTimer.singleShot(500, self._start_country_ip_resolution)
|
||
self._pending_subscription_additions.pop(subscription.id, None)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
self.subscription_update_finished.emit(result)
|
||
|
||
def _on_subscription_update_failed(self, worker_subscription, message: str) -> None:
|
||
self._log(f"[subscription] {worker_subscription.name or 'новая'}: {message}")
|
||
subscription = self.get_subscription(worker_subscription.id)
|
||
if subscription is not None:
|
||
result = mark_subscription_failure(subscription, message)
|
||
self.subscriptions_changed.emit(self.state.subscriptions)
|
||
self.save()
|
||
else:
|
||
self._pending_subscription_additions.pop(worker_subscription.id, None)
|
||
result = SubscriptionUpdateResult(
|
||
subscription_id=worker_subscription.id,
|
||
success=False,
|
||
message=message,
|
||
)
|
||
self.subscription_update_finished.emit(result)
|
||
|
||
def _on_subscription_worker_finished(self, subscription_id: str, worker) -> None:
|
||
if self._subscription_workers.get(subscription_id) is worker:
|
||
self._subscription_workers.pop(subscription_id, None)
|
||
self._subscription_check_ids.discard(subscription_id)
|
||
worker.deleteLater()
|
||
self._drain_subscription_update_queue()
|
||
self._maybe_finish_startup_connect()
|
||
|
||
def _check_due_subscriptions(self) -> None:
|
||
# AC3: global toggle is an outer AND-gate on top of per-subscription
|
||
# auto_update/backoff (subscription_due semantics unchanged).
|
||
if self.locked or not self.state.settings.subscriptions_auto_update:
|
||
return
|
||
for subscription in self.state.subscriptions:
|
||
if subscription_due(subscription):
|
||
self._enqueue_subscription_update(
|
||
subscription, "auto", pending_add=False, apply_result=True
|
||
)
|
||
|
||
def _startup_subscription_check_enabled(self) -> bool:
|
||
"""AC4: gate for the one-shot startup check (load() and startup unlock)."""
|
||
settings = self.state.settings
|
||
return bool(settings.subscriptions_auto_update and settings.subscriptions_check_on_startup)
|
||
|
||
def _apply_subscription_timer_interval(self) -> None:
|
||
"""AC5: subscription timer interval from settings (minutes, clamped 5..1440)."""
|
||
interval_ms = clamp_subscriptions_check_interval(
|
||
self.state.settings.subscriptions_check_interval_min
|
||
) * 60_000
|
||
if self._subscription_timer.interval() != interval_ms:
|
||
# setInterval on a running QTimer restarts it with the new period.
|
||
self._subscription_timer.setInterval(interval_ms)
|
||
|
||
def remove_nodes(self, node_ids: set[str]) -> None:
|
||
remove_nodes_operation(self, node_ids)
|
||
|
||
def update_node(self, node_id: str, updates: dict) -> bool:
|
||
return update_node_operation(self, node_id, updates)
|
||
|
||
def bulk_update_nodes(self, node_ids: set[str], operations: dict) -> int:
|
||
return bulk_update_nodes_operation(self, node_ids, operations)
|
||
|
||
def get_all_groups(self) -> list[str]:
|
||
return get_all_groups_operation(self)
|
||
|
||
def get_all_tags(self) -> list[str]:
|
||
return get_all_tags_operation(self)
|
||
|
||
def _migrate_sort_order(self) -> None:
|
||
if self.state.nodes and all(n.sort_order == 0 for n in self.state.nodes):
|
||
for i, node in enumerate(self.state.nodes):
|
||
node.sort_order = i + 1
|
||
self.schedule_save()
|
||
|
||
def reorder_nodes(self, node_id: str, direction: str) -> None:
|
||
reorder_nodes_operation(self, node_id, direction)
|
||
|
||
def set_selected_node(self, node_id: str, *, source: SelectionSource) -> None:
|
||
set_selected_node_operation(self, node_id, source=source)
|
||
|
||
def _set_connection_status(self, phase: str, message: str, level: str | None = None) -> None:
|
||
self.connection_status_changed.emit(phase, message)
|
||
if level is not None:
|
||
self.status.emit(level, message)
|
||
|
||
def _compute_connected_state(self) -> bool:
|
||
if self._active_session is not None and self._active_session.sidecar_kind == "amnezia":
|
||
return self.singbox.is_running and self.amnezia.is_running
|
||
if self._active_session is not None and self._active_session.hybrid:
|
||
return self.singbox.is_running and self.xray.is_running
|
||
if self._active_session is not None and self._active_session.sidecar_kind == "hysteria":
|
||
return self.singbox.is_running and self.hysteria.is_running
|
||
return self.singbox.is_running
|
||
|
||
def _refresh_connected_state(self) -> tuple[bool, bool]:
|
||
previous = self.connected
|
||
self.connected = self._compute_connected_state()
|
||
return previous, self.connected
|
||
|
||
# --- Rotation -----------------------------------------------------------
|
||
|
||
def xray_outbound_pool(self) -> XrayOutboundPool:
|
||
"""All Xray-compatible nodes loaded into the persistent data plane.
|
||
|
||
П3 (AC9/AC10): пул кэшируется по идентичности списка нод. Ключ ловит
|
||
замену списка, смену состава, переупорядочивание (sort_order) и замену
|
||
объекта ``node.outbound``; in-place правки содержимого outbound-словаря
|
||
покрываются явной инвалидацией по сигналу ``nodes_changed``.
|
||
"""
|
||
|
||
nodes = self.state.nodes
|
||
quarantine = self._xray_pool_quarantine_reasons()
|
||
key = (
|
||
id(nodes),
|
||
self._xray_pool_quarantine_version,
|
||
tuple((id(node), id(node.outbound), node.sort_order) for node in nodes),
|
||
)
|
||
if self._xray_outbound_pool_cache is None or self._xray_outbound_pool_cache_key != key:
|
||
self._xray_outbound_pool_cache = build_xray_outbound_pool(nodes, quarantine=quarantine)
|
||
self._xray_outbound_pool_cache_key = key
|
||
return self._xray_outbound_pool_cache
|
||
|
||
def _xray_pool_quarantine_reasons(self) -> dict[str, str]:
|
||
"""node_id -> причина для узлов, которые ядро отвергло и чей outbound не менялся."""
|
||
|
||
if not self._xray_pool_quarantine:
|
||
return {}
|
||
current = {node.id: id(node.outbound) for node in self.state.nodes}
|
||
stale = [
|
||
node_id
|
||
for node_id, (outbound_id, _reason) in self._xray_pool_quarantine.items()
|
||
if current.get(node_id) != outbound_id
|
||
]
|
||
for node_id in stale:
|
||
del self._xray_pool_quarantine[node_id]
|
||
if stale:
|
||
self._xray_pool_quarantine_version += 1
|
||
return {node_id: reason for node_id, (_outbound_id, reason) in self._xray_pool_quarantine.items()}
|
||
|
||
def _rejected_xray_tags(self) -> list[str]:
|
||
"""Теги, на которых ядро остановило разбор конфига при последнем запуске."""
|
||
|
||
tags = getattr(self.xray, "rejected_outbound_tags", None)
|
||
return list(tags) if isinstance(tags, (list, tuple)) else []
|
||
|
||
def _pool_candidate_nodes(self, selected: Node | None) -> list[Node]:
|
||
"""Узлы для сборки пула: без отвергнутых ядром, кроме выбранного.
|
||
|
||
Выбранный остаётся, чтобы запуск назвал причину отказа, а не молча
|
||
подменил сервер.
|
||
"""
|
||
|
||
reasons = self._xray_pool_quarantine_reasons()
|
||
if not reasons:
|
||
return self.state.nodes
|
||
keep = selected.id if selected is not None else ""
|
||
return [node for node in self.state.nodes if node.id not in reasons or node.id == keep]
|
||
|
||
def _quarantine_pruned_xray_outbounds(
|
||
self,
|
||
plan: SingboxRuntimePlan,
|
||
pruned_tags: list[str],
|
||
*,
|
||
notify: bool = True,
|
||
) -> str:
|
||
"""Занести отвергнутые ядром узлы в карантин и убрать их из плана.
|
||
|
||
Возвращает текст ошибки, если отвергнут выбранный сервер, иначе пустую строку.
|
||
``notify=False`` — запуск всё равно провалился и пользователь получит
|
||
ошибку ядра: отдельное предупреждение о пуле было бы шумом.
|
||
"""
|
||
|
||
sidecar = plan.xray_sidecar
|
||
tags = sidecar.outbound_pool_tags if sidecar is not None and sidecar.outbound_pool_tags else {}
|
||
by_tag = {tag: node_id for node_id, tag in tags.items()}
|
||
nodes_by_id = {node.id: node for node in self.state.nodes}
|
||
names: list[str] = []
|
||
selected_rejected = ""
|
||
for tag in pruned_tags:
|
||
node_id = by_tag.get(tag)
|
||
node = nodes_by_id.get(node_id or "")
|
||
if node is None:
|
||
continue
|
||
reason = "ядро Xray не смогло собрать этот сервер"
|
||
self._xray_pool_quarantine[node.id] = (id(node.outbound), reason)
|
||
names.append(node.name or node.id)
|
||
tags.pop(node.id, None)
|
||
if plan.selector_tags is not None:
|
||
plan.selector_tags.pop(node.id, None)
|
||
if tag == plan.selected_outbound_tag:
|
||
selected_rejected = node.name or node.id
|
||
if names:
|
||
self._xray_pool_quarantine_version += 1
|
||
self._invalidate_xray_outbound_pool_cache()
|
||
self._invalidate_plan_facts_cache()
|
||
shown = ", ".join(names[:5]) + (f" и ещё {len(names) - 5}" if len(names) > 5 else "")
|
||
self._log(f"[xray-pool] ядро Xray не приняло серверы, они убраны из пула: {shown}")
|
||
if notify and not selected_rejected:
|
||
self.status.emit("warning", f"Xray не принял серверов: {len(names)}. Они убраны из пула.")
|
||
if selected_rejected:
|
||
return (
|
||
f"Xray не принял выбранный сервер «{selected_rejected}»: его настройки не подходят "
|
||
"текущему ядру. Выберите другой сервер или обновите подписку."
|
||
)
|
||
return ""
|
||
|
||
def _invalidate_xray_outbound_pool_cache(self) -> None:
|
||
self._xray_outbound_pool_cache = None
|
||
self._xray_outbound_pool_cache_key = None
|
||
|
||
def _hot_switch_precheck(self) -> HotSwitchPlan | None:
|
||
"""Чистые (in-memory) проверки применимости горячего свитча (A3).
|
||
|
||
Никакого I/O: control-plane вызовы выполняются воркер-шагами в
|
||
``_hot_switch_selected_node_steps``. ``None`` означает фолбэк в очередь
|
||
переходов (``_request_transition``).
|
||
"""
|
||
|
||
node = self.selected_node
|
||
session = self._active_session
|
||
if getattr(self, "_transition_active", False) is True:
|
||
# Полный переход уже идёт: запрос сливается в очередь переходов,
|
||
# а не выполняет control-plane параллельно с ним.
|
||
self._log("[core-switch] transition in progress; request queued")
|
||
return None
|
||
if node is None:
|
||
self._log("[core-switch] fallback: selected node is missing")
|
||
return None
|
||
if session is None:
|
||
self._log("[core-switch] fallback: active session is missing")
|
||
return None
|
||
if not self.connected:
|
||
self._log("[core-switch] fallback: controller is not connected")
|
||
return None
|
||
tags = session.outbound_pool_tags or {}
|
||
tag = tags.get(node.id, "")
|
||
if not tag:
|
||
# Not a failure: a hot switch only re-points the running core at a
|
||
# member it already loaded. A sidecar session (Hysteria, AWG) runs
|
||
# exactly one server and has no pool, and a node outside the
|
||
# running pool needs different processes. Changing servers then
|
||
# requires the planned full transition, which replaces the
|
||
# processes and therefore ends in-flight connections.
|
||
if session.sidecar_kind and not tags:
|
||
reason = f"{session.sidecar_kind} sidecar session serves one server and has no switchable pool"
|
||
else:
|
||
reason = f"node is outside the running {session.active_core} pool ({len(tags)} tags)"
|
||
self._log(f"[core-switch] full transition to node {node.id}: {reason}")
|
||
return None
|
||
control_core = "xray" if session.hybrid else session.active_core
|
||
if control_core != "singbox" and not session.hybrid:
|
||
# Xray's RoutingService only changes the balancer choice for new
|
||
# connections. Without a sing-box selector in front there is no
|
||
# API capable of invalidating the existing TCP/UDP generation, so
|
||
# use the normal process transition instead of reporting a false
|
||
# successful cut-over.
|
||
self._log("[core-switch] fallback: Xray cannot interrupt existing connections")
|
||
return None
|
||
if session.tun_mode and not session.hybrid:
|
||
# Переключение selector'а «родного» sing-box под TUN давало
|
||
# «подключено» без TCP-трафика через туннель (замер на win10,
|
||
# 2026-10-14; и до правила Zapret на весь набор). Полный переход
|
||
# здесь стоит ~0,9 с и проверен трафиком.
|
||
self._log("[core-switch] full transition: native sing-box TUN selector switch is not trusted")
|
||
return None
|
||
if not self.bypass.ready_for(node):
|
||
self._log(f"[core-switch] full transition: Zapret rule for node {node.id} is not applied")
|
||
return None
|
||
return HotSwitchPlan(
|
||
node=node,
|
||
session=session,
|
||
tags=dict(tags),
|
||
tag=tag,
|
||
control_core=control_core,
|
||
previous_tag=tags.get(session.node_id or "", ""),
|
||
)
|
||
|
||
def _try_hot_switch_selected_node(self) -> bool:
|
||
"""Request a hot switch of the core's proxy transport (data plane stays alive).
|
||
|
||
П2 (AC5/AC8): чистые проверки выполняются синхронно в GUI-потоке,
|
||
control-plane I/O — шагами ``run_in_worker`` через TransitionRunner.
|
||
``True`` = «горячий путь принят» (свитч запущен либо поставлен за уже
|
||
идущим); ``False`` = вызывающий обязан идти через очередь переходов.
|
||
Отказ control-plane после принятия сам уводит в ``_request_transition``
|
||
(см. ``_on_hot_switch_runner_finished``).
|
||
"""
|
||
|
||
plan = self._hot_switch_precheck()
|
||
if plan is None:
|
||
return False
|
||
self._hot_switch_generation += 1
|
||
if self._hot_switch_runner is not None:
|
||
# AC8: сериализация — не больше одного control-plane вызова в
|
||
# полёте. Текущий runner устареет по generation (его результат не
|
||
# применится), а этот запрос диспетчеризуется по его завершении.
|
||
self._hot_switch_pending = True
|
||
return True
|
||
self._start_hot_switch_runner(plan)
|
||
return True
|
||
|
||
def _start_hot_switch_runner(self, plan: HotSwitchPlan) -> None:
|
||
generation = self._hot_switch_generation
|
||
transition_generation = self._transition_generation
|
||
runner = TransitionRunner(
|
||
self._hot_switch_selected_node_steps(plan),
|
||
# Устаревание (AC8): новый свитч, любой новый полный переход или
|
||
# потеря соединения отменяют текущий свитч до следующего шага;
|
||
# его результат (сессия/статус) не применяется.
|
||
is_current=lambda: (
|
||
self._hot_switch_generation == generation
|
||
and self._transition_generation == transition_generation
|
||
and self.connected
|
||
),
|
||
on_finished=self._on_hot_switch_runner_finished,
|
||
)
|
||
self._hot_switch_runner = runner
|
||
runner.start()
|
||
|
||
def _on_hot_switch_runner_finished(self, runner: TransitionRunner) -> None:
|
||
if self._hot_switch_runner is runner:
|
||
self._hot_switch_runner = None
|
||
runner.deleteLater()
|
||
_log_transition_profile(self, "core-switch", runner)
|
||
pending = self._hot_switch_pending
|
||
self._hot_switch_pending = False
|
||
if runner.cancelled:
|
||
if pending:
|
||
self._dispatch_hot_switch_request()
|
||
return
|
||
if runner.error is not None:
|
||
self._log(f"[core-switch] failed with error: {runner.error!r}")
|
||
ok = bool(runner.result) and runner.error is None
|
||
if not ok:
|
||
# AC7/C2: отказ control-plane честно падает в очередь переходов.
|
||
self._request_transition("node switched")
|
||
return
|
||
if pending:
|
||
self._dispatch_hot_switch_request()
|
||
|
||
def _dispatch_hot_switch_request(self) -> None:
|
||
"""Выполнить отложенный (superseded) свитч для актуально выбранной ноды."""
|
||
|
||
if not self.connected or not self._desired_connected:
|
||
return
|
||
node = self.selected_node
|
||
session = self._active_session
|
||
if node is not None and session is not None and session.node_id == node.id:
|
||
return # обгоняющий запрос уже удовлетворён завершившимся свитчем
|
||
if self._try_hot_switch_selected_node():
|
||
return
|
||
self._request_transition("node switched")
|
||
|
||
def _cancel_hot_switch_runner(self) -> None:
|
||
self._hot_switch_pending = False
|
||
runner = self._hot_switch_runner
|
||
if runner is not None:
|
||
runner.cancel()
|
||
|
||
def _apply_core_outbound_tag_steps(
|
||
self,
|
||
core: str,
|
||
outbound_tag: str,
|
||
*,
|
||
startup: bool = False,
|
||
) -> TransitionSteps:
|
||
"""Control-plane вызов на воркере (AC5); лог — в GUI-потоке; без pump (AC6).
|
||
|
||
Параметры (порты, путь к xray) снимаются в GUI-потоке, воркеру уходит
|
||
чистый callable без обращения к состоянию контроллера. ``startup`` —
|
||
холодный старт sing-box: Clash API ещё может не слушать, попытки
|
||
повторяются до дедлайна, пауза между ними — таймер, не GUI-поток.
|
||
"""
|
||
|
||
if core == "singbox" and startup:
|
||
ok, output = yield from select_singbox_outbound_when_ready_steps(
|
||
self._singbox_clash_api_port,
|
||
SINGBOX_SELECTOR_TAG,
|
||
outbound_tag,
|
||
)
|
||
if not ok:
|
||
self._log(f"[core-switch] {core} control plane rejected {outbound_tag}: {output}")
|
||
return bool(ok)
|
||
if core == "singbox":
|
||
api_port = self._singbox_clash_api_port
|
||
call = lambda: select_singbox_outbound(api_port, SINGBOX_SELECTOR_TAG, outbound_tag) # noqa: E731
|
||
else:
|
||
xray_path = getattr(self.xray, "_exe_path", None) or self.state.settings.xray_path
|
||
api_port = self._xray_api_port
|
||
call = lambda: apply_balancer_override( # noqa: E731
|
||
xray_path,
|
||
api_port,
|
||
XRAY_BALANCER_TAG,
|
||
outbound_tag,
|
||
pump=False,
|
||
)
|
||
ok, output = yield run_in_worker(call)
|
||
if not ok:
|
||
self._log(f"[core-switch] {core} control plane rejected {outbound_tag}: {output}")
|
||
return bool(ok)
|
||
|
||
def _hot_switch_selected_node_steps(self, plan: HotSwitchPlan) -> TransitionSteps:
|
||
"""Switch only the core's proxy transport, preserving the live data plane."""
|
||
|
||
started_at = time.perf_counter()
|
||
node, session, tags, tag = plan.node, plan.session, plan.tags, plan.tag
|
||
# Сервер из правила набора (не тот, под который оно строилось): его
|
||
# свежий DNS обязан попасть в правило winws2, иначе — полный переход.
|
||
verify = getattr(self.bypass, "verify_switch_target_steps", None)
|
||
if verify is not None and not (yield from verify(node)):
|
||
return False
|
||
if not (yield from self._apply_core_outbound_tag_steps(plan.control_core, tag)):
|
||
return False
|
||
|
||
hybrid_relay_selected_tag = session.hybrid_relay_selected_tag
|
||
if session.hybrid:
|
||
relay_tags = session.hybrid_relay_selector_tags
|
||
if len(relay_tags) < 2:
|
||
self._log("[core-switch] fallback: hybrid relay generations are unavailable")
|
||
if plan.previous_tag and plan.previous_tag != tag:
|
||
yield from self._apply_core_outbound_tag_steps("xray", plan.previous_tag)
|
||
return False
|
||
hybrid_relay_selected_tag = next(
|
||
(relay_tag for relay_tag in relay_tags if relay_tag != session.hybrid_relay_selected_tag),
|
||
relay_tags[0],
|
||
)
|
||
if not (yield from self._apply_core_outbound_tag_steps("singbox", hybrid_relay_selected_tag)):
|
||
self._log("[core-switch] hybrid cut-over rejected; restoring previous Xray outbound")
|
||
if plan.previous_tag and plan.previous_tag != tag:
|
||
yield from self._apply_core_outbound_tag_steps("xray", plan.previous_tag)
|
||
return False
|
||
|
||
# AC7: коммит сессии/GUI-статуса — только после подтверждения ядром.
|
||
self._capture_hot_switched_session(
|
||
node,
|
||
session,
|
||
tags,
|
||
tag,
|
||
hybrid_relay_selected_tag=hybrid_relay_selected_tag,
|
||
)
|
||
self._log(f"[core-switch-perf] total={(time.perf_counter() - started_at) * 1000:.1f}ms")
|
||
return True
|
||
|
||
def _capture_hot_switched_session(
|
||
self,
|
||
node: Node,
|
||
session: ActiveSessionSnapshot,
|
||
tags: dict[str, str],
|
||
tag: str,
|
||
*,
|
||
hybrid_relay_selected_tag: str = "",
|
||
) -> None:
|
||
"""Commit GUI/session state only after the core accepted the cut-over."""
|
||
|
||
node.last_used_at = utc_now_iso()
|
||
self._capture_active_session(
|
||
node,
|
||
tun=session.tun_mode,
|
||
core=session.active_core,
|
||
api_port=session.api_port,
|
||
hybrid=session.hybrid,
|
||
sidecar_kind=session.sidecar_kind,
|
||
socks_port=session.socks_port,
|
||
http_port=session.http_port,
|
||
xray_inbound_tags=session.xray_inbound_tags,
|
||
sidecar_relay_port=session.sidecar_relay_port,
|
||
protect_ss_port=session.protect_ss_port,
|
||
protect_ss_password=session.protect_ss_password,
|
||
ping_host=node.server,
|
||
ping_port=node.port,
|
||
outbound_pool_tags=tags,
|
||
hybrid_relay_selector_tags=session.hybrid_relay_selector_tags,
|
||
hybrid_relay_selected_tag=hybrid_relay_selected_tag,
|
||
)
|
||
self._set_connection_status("running", f"Переключено: {node.name}", level="success")
|
||
# П4: авто-переключение, завершившееся горячим свитчем (без reconnect),
|
||
# тоже обязано снять флаг «переход в процессе».
|
||
self._auto_switch_transitioning = False
|
||
self._auto_switch_link_down_since = 0.0
|
||
# Воркер метрик переживает горячий свитч — перенацелить его TCP-пинг,
|
||
# иначе детектор мёртвого сервера продолжит мерить прежнюю ноду.
|
||
worker = self._metrics_worker
|
||
if worker is not None:
|
||
worker.set_ping_target(
|
||
node.server,
|
||
int(node.port),
|
||
transport_kind_for_node(node),
|
||
)
|
||
self._log(f"[core-switch] {session.active_core} -> {node.name} ({tag})")
|
||
|
||
def is_rotation_supported(self, settings: AppSettings | None = None) -> bool:
|
||
"""Ротация опирается на пул outbound'ов, загруженный в работающее ядро."""
|
||
|
||
settings = settings or self.state.settings
|
||
return bool(settings.rotation_enabled)
|
||
|
||
def _rotation_available_ids(self) -> set[str] | None:
|
||
"""Ноды, которые запущенное ядро реально держит в своём пуле.
|
||
|
||
Переключение идёт по тегу, полученному ядром при старте, поэтому предлагать
|
||
ноду вне этого набора нельзя: свитч просто не состоится.
|
||
"""
|
||
|
||
session = self._active_session
|
||
if session is None:
|
||
return None
|
||
return set(session.outbound_pool_tags or {})
|
||
|
||
def rotation_plan(self, settings: AppSettings | None = None) -> RotationPlan | None:
|
||
settings = settings or self.state.settings
|
||
if not self.is_rotation_supported(settings):
|
||
return None
|
||
plan = build_rotation_plan(
|
||
settings,
|
||
self.state.nodes,
|
||
self.state.selected_node_id,
|
||
available_ids=self._rotation_available_ids(),
|
||
)
|
||
if plan is not None and plan.truncated:
|
||
# Метод вызывается часто (в т.ч. при расчёте сигнатур) — логируем факт
|
||
# усечения один раз на каждый новый размер пула, а не на каждый вызов.
|
||
mark = f"{plan.candidates}->{len(plan.nodes)}"
|
||
if mark != self._rotation_pool_logged:
|
||
self._rotation_pool_logged = mark
|
||
self._log(
|
||
f"[rotation] пул усечён: подходящих серверов {plan.candidates}, "
|
||
f"в ротации {len(plan.nodes)} (лимит настройки)"
|
||
)
|
||
return plan
|
||
|
||
@staticmethod
|
||
def _rotation_settings_signature(settings: AppSettings) -> tuple:
|
||
return (
|
||
bool(settings.rotation_enabled),
|
||
str(settings.rotation_mode),
|
||
int(settings.rotation_interval_sec),
|
||
int(settings.rotation_jitter_pct),
|
||
)
|
||
|
||
def _sync_rotation_timer(self) -> None:
|
||
plan = self.rotation_plan()
|
||
if plan is None or not self.connected:
|
||
self._rotation_timer.stop()
|
||
if self._rotation_running:
|
||
self._rotation_running = False
|
||
self._log("[rotation] остановлена")
|
||
return
|
||
if not self._rotation_running:
|
||
self._rotation_running = True
|
||
self._log(
|
||
f"[rotation] запущена: серверов в ротации {len(plan.nodes)}, "
|
||
f"режим {self.state.settings.rotation_mode}"
|
||
)
|
||
self._rotation_timer.start(rotation_interval_ms(self.state.settings, self._rotation_rng))
|
||
|
||
def _on_rotation_tick(self) -> None:
|
||
# Собственный слот single-shot таймера: isActive() здесь уже False, поэтому
|
||
# состояние «ротация идёт» отслеживается флагом, а не таймером — иначе каждый
|
||
# тик заново логировался бы как запуск.
|
||
self.rotate_now()
|
||
self._sync_rotation_timer()
|
||
|
||
def rotate_now(self) -> bool:
|
||
"""Переключиться на следующий сервер ротации без перезапуска ядра."""
|
||
|
||
plan = self.rotation_plan()
|
||
if plan is None or not self.connected:
|
||
return False
|
||
node = pick_next_node(
|
||
plan,
|
||
self.state.selected_node_id,
|
||
self.state.settings.rotation_mode,
|
||
self._rotation_rng,
|
||
)
|
||
if node is None or node.id == self.state.selected_node_id:
|
||
return False
|
||
|
||
self._log(f"[rotation] переключение -> {node.name}")
|
||
# Штатный путь смены сервера: он сам делает горячий свитч по тегу, который
|
||
# ядро получило при запуске, а при неудаче честно переподключается.
|
||
self.set_selected_node(node.id, source=SelectionSource.ROTATION)
|
||
self.status.emit("info", f"Ротация: {node.name}")
|
||
return True
|
||
|
||
def _reset_auto_switch_state(self, *, reset_cooldown: bool = False, reset_cycle: bool = True) -> None:
|
||
self._auto_switch_link_down_since = 0.0
|
||
if reset_cycle:
|
||
self._auto_switch_cycle_attempts = 0
|
||
self._auto_switch_exhausted = False
|
||
if reset_cooldown:
|
||
self._auto_switch_last_switch = 0.0
|
||
|
||
def _handle_auto_switch_setting_change(self, old_enabled: bool, new_enabled: bool) -> None:
|
||
"""Apply the explicit auto-switch toggle to the session guard."""
|
||
|
||
if bool(old_enabled) == bool(new_enabled):
|
||
return
|
||
if new_enabled:
|
||
# Re-enabling the setting is the explicit user action that releases
|
||
# the manual-selection hold. A threshold/cooldown edit alone must
|
||
# not silently override a manually chosen server.
|
||
self._auto_switch_manual_hold = False
|
||
self._smart_switch_manual_at = 0.0
|
||
self._reset_auto_switch_state(reset_cooldown=True, reset_cycle=True)
|
||
if self.connected:
|
||
begin_auto_switch_warmup(self, self.selected_node)
|
||
return
|
||
self._reset_auto_switch_state(reset_cooldown=False, reset_cycle=True)
|
||
|
||
def _cleanup_connection_runtime_state(
|
||
self,
|
||
*,
|
||
end_traffic_session: bool,
|
||
reset_auto_switch_cycle: bool,
|
||
reset_auto_switch_cooldown: bool,
|
||
) -> None:
|
||
cleanup_connection_runtime_state_operation(
|
||
self,
|
||
end_traffic_session=end_traffic_session,
|
||
reset_auto_switch_cycle=reset_auto_switch_cycle,
|
||
reset_auto_switch_cooldown=reset_auto_switch_cooldown,
|
||
)
|
||
|
||
def _stop_active_connection_processes(self, *, disable_proxy: bool) -> bool:
|
||
"""Синхронный драйвер — только shutdown."""
|
||
return stop_active_connection_processes_operation(self, disable_proxy=disable_proxy)
|
||
|
||
def _stop_active_connection_processes_steps(self, *, disable_proxy: bool) -> TransitionSteps:
|
||
return (yield from stop_active_connection_processes_steps_operation(self, disable_proxy=disable_proxy))
|
||
|
||
def _handle_unexpected_disconnect(self) -> None:
|
||
"""Из обработчиков сигналов: fail-closed без ожидания в GUI-потоке."""
|
||
handle_unexpected_disconnect_operation(self)
|
||
|
||
def _handle_unexpected_disconnect_steps(self) -> TransitionSteps:
|
||
"""Внутри перехода: остановка процессов шагами."""
|
||
return (yield from handle_unexpected_disconnect_steps_operation(self))
|
||
|
||
def connect_selected(self, allow_during_reconnect: bool = False) -> bool:
|
||
"""Синхронный драйвер (тесты). Переходы — через координатор."""
|
||
return connect_selected_operation(self, allow_during_reconnect=allow_during_reconnect)
|
||
|
||
def _connect_selected_steps(self, allow_during_reconnect: bool = False) -> TransitionSteps:
|
||
return (yield from connect_selected_steps_operation(self, allow_during_reconnect=allow_during_reconnect))
|
||
|
||
def disconnect_current(self, disable_proxy: bool = True, emit_status: bool = True) -> bool:
|
||
"""Синхронный драйвер: shutdown и обновление ядра Xray. Клики — через координатор."""
|
||
return disconnect_current_operation(self, disable_proxy=disable_proxy, emit_status=emit_status)
|
||
|
||
def _disconnect_current_steps(self, disable_proxy: bool = True, emit_status: bool = True) -> TransitionSteps:
|
||
return (yield from disconnect_current_steps_operation(
|
||
self, disable_proxy=disable_proxy, emit_status=emit_status,
|
||
))
|
||
|
||
def _restart_proxy_core(self, reason: str) -> bool:
|
||
return restart_singbox_proxy_runtime_operation(self, reason)
|
||
|
||
def _restart_proxy_core_steps(self, reason: str) -> TransitionSteps:
|
||
return (yield from restart_singbox_proxy_runtime_steps_operation(self, reason))
|
||
|
||
def _restart_singbox_runtime(self, reason: str) -> bool:
|
||
return restart_singbox_runtime_operation(self, reason)
|
||
|
||
@property
|
||
def traffic_history(self) -> TrafficHistoryStorage:
|
||
return self._traffic_history
|
||
|
||
def toggle_connection(self) -> None:
|
||
current_target = self._desired_connected if (self._transition_active or self._transition_pending) else self.connected
|
||
self._desired_connected = not current_target
|
||
# Ручное действие отменяет отложенный возврат после неудачной замены.
|
||
self._hysteria_return_episode = 0
|
||
if not self._desired_connected:
|
||
self._clear_pending_transport_selection()
|
||
self._hysteria_recovery_active = False
|
||
self._request_transition("toggle connection")
|
||
|
||
def switch_next_node(self, *, source: SelectionSource) -> None:
|
||
if not self.state.nodes:
|
||
return
|
||
current_id = self.state.selected_node_id
|
||
index = 0
|
||
if current_id:
|
||
for idx, node in enumerate(self.state.nodes):
|
||
if node.id == current_id:
|
||
index = idx
|
||
break
|
||
index = (index + 1) % len(self.state.nodes)
|
||
self.set_selected_node(self.state.nodes[index].id, source=source)
|
||
|
||
def update_routing(self, routing: RoutingSettings) -> None:
|
||
if routing.mode not in ROUTING_MODES:
|
||
routing.mode = "rule"
|
||
self.state.routing = routing
|
||
self.routing_changed.emit(self.state.routing)
|
||
self.schedule_save()
|
||
|
||
if self.connected or self._desired_connected:
|
||
if not self.is_tun2socks_mode():
|
||
return
|
||
self._request_transition("routing changed")
|
||
|
||
def repair_launch_on_startup(self) -> None:
|
||
"""Сверить задание автозапуска с настройкой и текущим путём приложения."""
|
||
import threading
|
||
|
||
threading.Thread(
|
||
target=windows_startup.repair,
|
||
args=(self.state.settings.launch_on_startup,),
|
||
name="autostart-repair",
|
||
daemon=True,
|
||
).start()
|
||
|
||
def update_settings(self, settings: AppSettings) -> None:
|
||
settings.proxy_engine = "singbox"
|
||
settings.tun_engine = "singbox"
|
||
old_settings = self.state.settings
|
||
old_launch = old_settings.launch_on_startup
|
||
old_tun = old_settings.tun_mode
|
||
old_proxy_engine = old_settings.proxy_engine
|
||
old_tun_engine = old_settings.tun_engine
|
||
old_rotation = self._rotation_settings_signature(old_settings)
|
||
old_auto_switch_enabled = old_settings.auto_switch_enabled
|
||
self.state.settings = settings
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.schedule_save()
|
||
self._apply_subscription_timer_interval()
|
||
|
||
self._handle_auto_switch_setting_change(
|
||
old_auto_switch_enabled,
|
||
settings.auto_switch_enabled,
|
||
)
|
||
if not (settings.auto_switch_enabled and settings.auto_switch_low_speed_enabled):
|
||
cancel_smart_check(self, "авто-переключение при низкой скорости выключено")
|
||
|
||
if old_rotation != self._rotation_settings_signature(settings):
|
||
# Смена интервала/режима не трогает конфиг — достаточно перепланировать таймер.
|
||
self._rotation_timer.stop()
|
||
self._sync_rotation_timer()
|
||
|
||
if old_launch != settings.launch_on_startup:
|
||
try:
|
||
windows_startup.apply(settings.launch_on_startup)
|
||
except Exception as exc:
|
||
# Переключатель обязан показывать правду: настройка, которую
|
||
# Windows не приняла, возвращается в прежнее положение.
|
||
settings.launch_on_startup = old_launch
|
||
self.settings_changed.emit(self.state.settings)
|
||
self.status.emit("error", f"Не удалось настроить автозапуск: {exc}")
|
||
|
||
if self.connected or self._desired_connected:
|
||
if old_tun != settings.tun_mode:
|
||
self._desired_connected = True
|
||
self._request_transition("TUN mode toggled")
|
||
return
|
||
if settings.tun_mode and old_tun_engine != settings.tun_engine:
|
||
self._desired_connected = True
|
||
self._request_transition("TUN engine changed")
|
||
return
|
||
if not settings.tun_mode and old_proxy_engine != settings.proxy_engine:
|
||
self._desired_connected = True
|
||
self._request_transition("proxy engine changed")
|
||
return
|
||
self._request_transition("settings changed")
|
||
|
||
def ping_nodes(self, node_ids: set[str] | None = None) -> None:
|
||
ping_nodes_operation(self, node_ids)
|
||
|
||
def cancel_ping(self) -> bool:
|
||
return cancel_ping_operation(self)
|
||
|
||
def speed_test_nodes(self, node_ids: set[str] | None = None) -> bool:
|
||
return speed_test_nodes_operation(self, node_ids)
|
||
|
||
def cancel_speed_test(self) -> bool:
|
||
return cancel_speed_test_operation(self)
|
||
|
||
def get_fastest_alive_node(self) -> Node | None:
|
||
return get_fastest_alive_node_operation(self)
|
||
|
||
def test_connectivity(self, url: str | None = None) -> None:
|
||
test_connectivity_operation(self, url)
|
||
|
||
def run_xray_core_update(self, apply_update: bool, silent: bool = False) -> None:
|
||
run_xray_core_update_operation(self, apply_update, silent=silent)
|
||
|
||
def _start_metrics_worker(self) -> None:
|
||
start_metrics_worker_operation(self)
|
||
|
||
def _stop_metrics_worker(self) -> None:
|
||
stop_metrics_worker_operation(self)
|
||
|
||
def set_master_password(self, password: str) -> None:
|
||
password_hash, salt = create_password_hash(password)
|
||
self.state.security.enabled = True
|
||
self.state.security.password_hash = password_hash
|
||
self.state.security.salt = salt
|
||
self.save()
|
||
|
||
def disable_master_password(self) -> None:
|
||
self.state.security.enabled = False
|
||
self.state.security.password_hash = ""
|
||
self.state.security.salt = ""
|
||
self.locked = False
|
||
self.lock_state_changed.emit(False)
|
||
self.save()
|
||
|
||
def unlock(self, password: str) -> bool:
|
||
if not self.state.security.enabled:
|
||
self.locked = False
|
||
self.lock_state_changed.emit(False)
|
||
if self._startup_subscription_check_enabled():
|
||
QTimer.singleShot(0, self._check_due_subscriptions)
|
||
return True
|
||
|
||
ok = verify_password(password, self.state.security.password_hash, self.state.security.salt)
|
||
if ok:
|
||
self.locked = False
|
||
self.lock_state_changed.emit(False)
|
||
if self._startup_subscription_check_enabled():
|
||
QTimer.singleShot(0, self._check_due_subscriptions)
|
||
return ok
|
||
|
||
def lock(self) -> None:
|
||
if not self.state.security.enabled:
|
||
return
|
||
self.locked = True
|
||
self.lock_state_changed.emit(True)
|
||
self._desired_connected = False
|
||
self._request_transition("locked")
|
||
|
||
def build_diagnostics(self) -> Path:
|
||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
output = LOG_DIR / f"diagnostics_{stamp}.zip"
|
||
return export_diagnostics(output, self.state, self.recent_logs,
|
||
runtime_errors=self.runtime_errors.snapshot(),
|
||
runtime=collect_runtime_diagnostics(self))
|
||
|
||
def auto_connect_if_needed(self) -> None:
|
||
if not self.state.settings.auto_connect_last or self.locked:
|
||
return
|
||
if self.selected_node is None and not self._can_connect_without_selected_node():
|
||
return
|
||
if (
|
||
self.state.settings.startup_connect_order == "after_subscriptions"
|
||
and self._begin_startup_connect_wait()
|
||
):
|
||
return
|
||
self._perform_auto_connect()
|
||
|
||
def resume_after_app_update(self, was_connected: bool) -> None:
|
||
"""Первый запуск после автообновления: вернуть состояние до перезапуска.
|
||
|
||
Обычное автоподключение смотрит только на настройку и включило бы VPN,
|
||
который пользователь выключил перед тихим перезапуском.
|
||
"""
|
||
if not was_connected:
|
||
self._log("[update] restart after app update: staying disconnected")
|
||
return
|
||
if self.state.settings.auto_connect_last:
|
||
# Обычный путь сохраняет порядок «подключаться после подписок».
|
||
self.auto_connect_if_needed()
|
||
return
|
||
if self.locked:
|
||
return
|
||
if self.selected_node is None and not self._can_connect_without_selected_node():
|
||
return
|
||
self._log("[update] restart after app update: restoring connection")
|
||
self._desired_connected = True
|
||
self._request_transition("app update resume")
|
||
|
||
def transition_busy(self) -> bool:
|
||
return bool(self._transition_active or self._transition_pending)
|
||
|
||
def _perform_auto_connect(self) -> None:
|
||
if not self.state.settings.auto_connect_last or self.locked:
|
||
return
|
||
if self.selected_node is not None or self._can_connect_without_selected_node():
|
||
self._desired_connected = True
|
||
self._request_transition("auto connect")
|
||
|
||
def _begin_startup_connect_wait(self) -> bool:
|
||
"""AC6: try deferring auto-connect until startup subscription refresh ends.
|
||
|
||
Returns True only when a startup check actually queued work; the armed
|
||
fallback timer (STARTUP_CONNECT_FALLBACK_MS <= 30 s) guarantees that a
|
||
failed or hanging subscription update never blocks auto-connect.
|
||
"""
|
||
if not self._startup_subscription_check_enabled():
|
||
return False
|
||
self._check_due_subscriptions()
|
||
if not self._subscription_workers and not self._subscription_update_queue:
|
||
return False # nothing to update -> degenerate to immediate connect
|
||
self._startup_connect_pending = True
|
||
self._startup_connect_timer.start()
|
||
return True
|
||
|
||
def _maybe_finish_startup_connect(self) -> None:
|
||
if (
|
||
self._startup_connect_pending
|
||
and not self._subscription_workers
|
||
and not self._subscription_update_queue
|
||
):
|
||
self._finish_startup_connect()
|
||
|
||
def _finish_startup_connect(self) -> None:
|
||
if not self._startup_connect_pending:
|
||
return
|
||
self._startup_connect_pending = False
|
||
self._startup_connect_timer.stop()
|
||
self._perform_auto_connect()
|
||
|
||
def _log(self, line: str) -> None:
|
||
"""Send a log line to the UI and write it to the log file."""
|
||
line = redact_server_addresses(line)
|
||
self.recent_logs.append(line)
|
||
if len(self.recent_logs) > 5000:
|
||
self.recent_logs = self.recent_logs[-5000:]
|
||
self._logger.info(line)
|
||
self.log_line.emit(line)
|
||
|
||
def _on_xray_log(self, line: str) -> None:
|
||
self._on_core_log("xray", line)
|
||
|
||
def _on_core_log(self, engine: str, line: str) -> None:
|
||
if "DNS_FALLBACK server=" in line:
|
||
self._report_dns_fallback(line)
|
||
health_log = engine == "hysteria" and re.match(
|
||
r"^\[hysteria\]\[attempt=\d+ stage=(?:health_check|destination)(?: |\])", line
|
||
)
|
||
# Optional endpoint health and per-destination refusals by an
|
||
# authenticated server are evidence, not a transport failure episode.
|
||
if is_deprecation_notice(line):
|
||
# Не сбой: ядро работает. Один раз на текст, остальное считаем.
|
||
if not self._core_notices.first_time(engine, line):
|
||
return
|
||
line += " (повторяется для каждого сервера пула, дальше скрыто)"
|
||
elif is_core_error_line(line) and not health_log:
|
||
self._record_core_failure(engine, "runtime", line)
|
||
context = self._core_log_contexts.get(engine)
|
||
if context is None:
|
||
context = RuntimeLogContext(
|
||
engine=engine,
|
||
role="core",
|
||
mode="tun" if self.state.settings.tun_mode else "proxy",
|
||
generation=int(self._transition_generation),
|
||
)
|
||
line = contextualize_runtime_log(line, context=context)
|
||
# In TUN mode, throttle noisy per-connection logs to prevent UI freeze
|
||
if self.state.settings.tun_mode and "accepted" in line:
|
||
self._tun_log_count = getattr(self, "_tun_log_count", 0) + 1
|
||
# Only log to file, skip UI — emit summary every 100 lines
|
||
self._logger.info(line)
|
||
self.recent_logs.append(line)
|
||
if len(self.recent_logs) > 5000:
|
||
self.recent_logs = self.recent_logs[-5000:]
|
||
if self._tun_log_count % 100 == 0:
|
||
self.log_line.emit(f"[tun] {self._tun_log_count} connections routed...")
|
||
return
|
||
self._log(line)
|
||
|
||
def _report_dns_fallback(self, line: str) -> None:
|
||
if "DNS_FALLBACK server=local-system-dns" in line:
|
||
level = 2
|
||
message = "DoH недоступен. Используется обычный DNS системы: запросы могут быть видны провайдеру."
|
||
elif "DNS_FALLBACK server=app-bootstrap-lkg" in line:
|
||
level = 2
|
||
message = (
|
||
"Защищённый DNS недоступен. Адрес сервера взят из запомненных: "
|
||
"если сервер переехал, подключение не состоится."
|
||
)
|
||
elif "DNS_FALLBACK server=bootstrap-dns" in line:
|
||
level = 1
|
||
message = "DNS через VPN недоступен. Используется резервное разрешение DNS с устройства."
|
||
else:
|
||
return
|
||
generation = self._transition_generation
|
||
previous = getattr(self, "_dns_warning_state", None)
|
||
if not isinstance(previous, tuple) or previous[0] != generation or level > previous[1]:
|
||
self._dns_warning_state = (generation, level)
|
||
self.status.emit("warning", message)
|
||
|
||
def _on_xray_error(self, message: str) -> None:
|
||
self._on_core_error("xray", message)
|
||
|
||
def _on_core_error(self, engine: str, message: str) -> None:
|
||
self._record_core_failure(engine, "operation", message)
|
||
context = self._core_log_contexts.get(engine)
|
||
if context is None:
|
||
context = RuntimeLogContext(
|
||
engine=engine,
|
||
role="core",
|
||
mode="tun" if self.state.settings.tun_mode else "proxy",
|
||
generation=int(self._transition_generation),
|
||
)
|
||
detailed = contextualize_runtime_log(message, context=context, error=True)
|
||
self._log(detailed)
|
||
self._set_connection_status("error", detailed, level="error")
|
||
|
||
def _record_core_failure(self, engine: str, stage: str, message: str) -> None:
|
||
journal = getattr(self, "runtime_errors", None)
|
||
if journal is None:
|
||
self.runtime_errors = journal = RuntimeErrorJournal()
|
||
context = self._core_log_contexts.get(engine)
|
||
journal.record(core_failure(
|
||
engine, stage, message,
|
||
session_generation=context.generation if context else 0,
|
||
# A shared log stream does not prove which pooled target emitted it.
|
||
# Never attribute late output to the currently selected node.
|
||
))
|
||
self._schedule_runtime_errors_emit()
|
||
|
||
def _schedule_runtime_errors_emit(self) -> None:
|
||
"""Снимок журнала в GUI не чаще раза в секунду.
|
||
|
||
Шумное ядро пишет ошибки десятками в секунду; копировать и
|
||
перерисовывать весь журнал на каждую строку незачем.
|
||
"""
|
||
timer = getattr(self, "_runtime_errors_emit_timer", None)
|
||
if timer is None:
|
||
timer = QTimer(self)
|
||
timer.setSingleShot(True)
|
||
timer.setInterval(1000)
|
||
timer.timeout.connect(lambda: self.runtime_errors_changed.emit(self.runtime_errors.snapshot()))
|
||
self._runtime_errors_emit_timer = timer
|
||
if not timer.isActive():
|
||
timer.start()
|
||
|
||
def _on_singbox_error(self, message: str) -> None:
|
||
self._on_core_error("sing-box", message)
|
||
|
||
def _on_core_stopped(self, core: str, exit_code: int) -> None:
|
||
for summary in self._core_notices.drain_summary(core):
|
||
self._log(summary)
|
||
self._log(f"[{core}] process stopped with code {exit_code}")
|
||
|
||
def _on_core_state_changed(self, _running: bool) -> None:
|
||
on_core_state_changed_operation(self, _running)
|
||
|
||
def _on_ping_measured(self, outcome) -> None:
|
||
on_ping_measured_operation(self, outcome)
|
||
|
||
def _on_ping_progress(self, done: int, total: int, finished: bool) -> None:
|
||
on_ping_progress_operation(self, done, total, finished)
|
||
|
||
def _on_speed_result(self, node_id: str, speed_mbps: float | None, is_alive: bool) -> None:
|
||
on_speed_result_operation(self, node_id, speed_mbps, is_alive)
|
||
|
||
def _on_services_result(self, node_id: str, verdicts: object) -> None:
|
||
on_services_result_operation(self, node_id, verdicts if isinstance(verdicts, dict) else {})
|
||
|
||
def _on_speed_progress(self, current: int, total: int) -> None:
|
||
on_speed_progress_operation(self, current, total)
|
||
|
||
def _on_speed_node_progress(self, node_id: str, percent: int) -> None:
|
||
on_speed_node_progress_operation(self, node_id, percent)
|
||
|
||
def _on_speed_complete(self) -> None:
|
||
on_speed_complete_operation(self)
|
||
|
||
def _on_connectivity_result(self, ok: bool, message: str, elapsed_ms: int | None) -> None:
|
||
on_connectivity_result_operation(self, ok, message, elapsed_ms)
|
||
|
||
def _on_live_metrics(self, payload: dict[str, object]) -> None:
|
||
on_live_metrics_operation(self, payload)
|
||
|
||
def _check_auto_switch(
|
||
self,
|
||
down_bps: float,
|
||
link_alive: bool | None = None,
|
||
*,
|
||
traffic_valid: bool = True,
|
||
) -> None:
|
||
check_auto_switch_operation(
|
||
self,
|
||
down_bps,
|
||
link_alive,
|
||
traffic_valid=traffic_valid,
|
||
)
|
||
|
||
def _get_next_node_for_auto_switch(self) -> Node | None:
|
||
return get_next_node_for_auto_switch_operation(self)
|
||
|
||
def _check_smart_switch(
|
||
self,
|
||
down_bps: float,
|
||
*,
|
||
traffic_valid: bool = True,
|
||
demand: dict[str, object] | None = None,
|
||
) -> None:
|
||
check_smart_switch_operation(
|
||
self,
|
||
down_bps,
|
||
traffic_valid=traffic_valid,
|
||
demand=demand,
|
||
)
|
||
|
||
# Слоты воркеров умной проверки: связь сигнал→метод QObject идёт очередью
|
||
# Qt, поэтому решение и set_selected_node выполняются в GUI-потоке.
|
||
def _on_smart_probe_measured(self, bps: object) -> None:
|
||
on_smart_probe_measured_operation(
|
||
self, self.sender(), float(bps) if isinstance(bps, (int, float)) else None,
|
||
)
|
||
|
||
def _on_smart_candidate_result(self, node_id: str, speed_mbps: object, is_alive: bool) -> None:
|
||
on_smart_candidate_result_operation(
|
||
self,
|
||
self.sender(),
|
||
node_id,
|
||
float(speed_mbps) if isinstance(speed_mbps, (int, float)) else None,
|
||
is_alive,
|
||
)
|
||
|
||
def _on_smart_candidates_done(self) -> None:
|
||
on_smart_candidates_done_operation(self, self.sender())
|
||
|
||
def _on_xray_update_worker_done(self, result: XrayCoreUpdateResult) -> None:
|
||
on_xray_update_worker_done_operation(self, result)
|
||
|
||
def _active_tunnel_addresses(self) -> set[str]:
|
||
"""IPs assigned to the active AWG/WireGuard sidecar's own tunnel.
|
||
|
||
These become the outbound IP once the handshake completes, which the
|
||
network monitor otherwise misreads as a real network change.
|
||
"""
|
||
plan = getattr(self, "_active_singbox_plan", None)
|
||
sidecar = getattr(plan, "amnezia_sidecar", None) if plan is not None else None
|
||
if sidecar is None:
|
||
return set()
|
||
addresses = ((sidecar.config or {}).get("endpoint") or {}).get("address") or []
|
||
return {str(item).split("/", 1)[0].strip() for item in addresses if str(item).strip()}
|
||
|
||
def _on_network_changed(self, old: str, new: str) -> None:
|
||
self._log(f"[network] changed: {old} -> {new}")
|
||
# TUN mode creates a virtual adapter which triggers network change —
|
||
# reconnecting would kill the TUN and cause an infinite loop
|
||
if self.state.settings.tun_mode:
|
||
self._log("[network] ignoring change in TUN mode")
|
||
return
|
||
# A proxy-mode AWG/WireGuard sidecar owns a virtual tunnel whose address
|
||
# becomes the outbound IP after a successful handshake. Reconnecting on
|
||
# that self-induced change re-runs the physical-interface resolve while
|
||
# the tunnel holds the default route, which fails and flaps the tunnel.
|
||
tunnel_addresses = self._active_tunnel_addresses()
|
||
if tunnel_addresses and (new in tunnel_addresses or old in tunnel_addresses):
|
||
self._log("[network] ignoring change caused by the tunnel's own address")
|
||
return
|
||
if (self.connected and self._desired_connected
|
||
and not self._transition_active and not self._disconnecting
|
||
and self.state.settings.reconnect_on_network_change):
|
||
self._desired_connected = True
|
||
self._request_transition("network changed")
|
||
|
||
def _hot_swap_node(self, reason: str) -> bool:
|
||
"""Handle node switch while TUN is active (синхронный совместимый путь)."""
|
||
return bool(run_steps_blocking(self._hot_swap_node_steps(reason)))
|
||
|
||
def _hot_swap_node_steps(self, reason: str) -> TransitionSteps:
|
||
"""Handle node switch while TUN is active."""
|
||
session = self._active_session
|
||
if session is None:
|
||
self._auto_switch_transitioning = False
|
||
return False
|
||
|
||
self._xray_api_port = session.api_port
|
||
self._protect_ss_port = session.protect_ss_port
|
||
self._protect_ss_password = session.protect_ss_password
|
||
|
||
try:
|
||
return (yield from restart_singbox_runtime_steps_operation(self, reason))
|
||
finally:
|
||
self._auto_switch_transitioning = False
|
||
|
||
def _reconnect(self, reason: str) -> bool:
|
||
return reconnect_operation(self, reason)
|
||
|
||
def _reconnect_steps(self, reason: str) -> TransitionSteps:
|
||
return (yield from reconnect_steps_operation(self, reason))
|
||
|
||
def export_backup(self, path: Path, passphrase: str = "") -> None:
|
||
self.storage.export_backup(path, passphrase)
|
||
|
||
def import_backup(self, path: Path, passphrase: str = "") -> None:
|
||
self.state = self.storage.import_backup(path, passphrase)
|
||
self.save()
|
||
self.nodes_changed.emit(self.state.nodes)
|
||
self.selection_changed.emit(self.selected_node)
|
||
self.routing_changed.emit(self.state.routing)
|
||
self.settings_changed.emit(self.state.settings)
|
||
|
||
def _check_auto_lock(self) -> None:
|
||
if not self.state.security.enabled:
|
||
return
|
||
if self.locked:
|
||
return
|
||
minutes = max(1, self.state.security.auto_lock_minutes)
|
||
if get_idle_seconds() >= minutes * 60:
|
||
self.lock()
|