loop-uh
8d6a05b8d8
v0.6.6 still failed AWG startup with "Physical interface for Amnezia UDP
transport not found" on the common Hyper-V vEthernet setup. Live probing on
real Windows showed the cause: GetAdaptersAddresses leaves FirstGatewayAddress
empty on working uplinks (Hyper-V vEthernet, some DHCP configs), so the
gateway-based selector rejected every adapter and returned None. (Interface
index, type, metric, addresses and DNS all parsed correctly — only the gateway
list was unreliable.)
Resolve the uplink with GetBestInterfaceEx instead: ask the OS which interface
reaches a destination — the authoritative default-route decision, independent
of the gateway list. Pass the AWG server's own IP so the physical uplink is
returned even while a tunnel holds the default route (WireGuard keeps a host
route to the server off the tunnel). The metric-based selector remains as a
fallback, now excluding tunnel/loopback by IfType instead of requiring a
gateway. Verified on real Windows: resolve_physical_uplink() -> (10, [dns]).
Also:
- Hysteria/Amnezia no longer raise a user-facing warning when the background
DoH health probes (1.1.1.1/8.8.8.8/9.9.9.9) fail on a censored exit: the
server handshake is already authenticated and real traffic flows, so the
probe failure is a false alarm. The diagnostic log is kept.
- AWG startup failures now reach the info bar. Previously _on_amnezia_failure
returned early when no amnezia session was committed yet, so a failed AWG
start surfaced nothing to the user ("just exits").
892 unit tests pass (offscreen). End-to-end AWG connect still needs on-device
confirmation on the Hyper-V machine.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HAjvZYzPW2yTtToKJXGdbS
790 lines
33 KiB
Python
790 lines
33 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import deque
|
||
from concurrent.futures import Future, ThreadPoolExecutor
|
||
from copy import deepcopy
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import time
|
||
from typing import Any
|
||
|
||
from PyQt6.QtCore import QObject, QProcess, QTimer, pyqtSignal
|
||
|
||
from ...constants import HYSTERIA_CONFIG_FILE, HYSTERIA_PATH_DEFAULT, PROXY_HOST, RUNTIME_DIR
|
||
from ...diagnostics.export import capture_runtime_config
|
||
from ..socks_probe import HTTPS_ENDPOINTS, probe_https
|
||
from ..health_check import BackgroundHealthCheck
|
||
from ..sidecar import wait_for_loopback_relay
|
||
from .runtime_contract import (
|
||
SECURITY_FAILURES,
|
||
HysteriaFailureCode,
|
||
classify_hysteria_failure,
|
||
)
|
||
from ...diagnostics.runtime_logging import RuntimeNodeIdentity, redact_runtime_log, strip_terminal_controls
|
||
from ...platform.windows.subprocess_utils import (
|
||
decode_output,
|
||
kill_processes_by_path,
|
||
sleep_with_events,
|
||
wait_for_qprocess_finished,
|
||
wait_for_qprocess_started,
|
||
)
|
||
|
||
|
||
_FUNCTIONAL_HTTPS_ENDPOINTS = HTTPS_ENDPOINTS
|
||
|
||
|
||
class HysteriaManager(QObject):
|
||
"""Run the unmodified official Hysteria client as a local SOCKS sidecar."""
|
||
|
||
started = pyqtSignal()
|
||
stopped = pyqtSignal(int)
|
||
log_received = pyqtSignal(str)
|
||
error = pyqtSignal(str)
|
||
warning = pyqtSignal(str)
|
||
failure = pyqtSignal(str, str, int)
|
||
state_changed = pyqtSignal(bool)
|
||
|
||
def __init__(self, parent: QObject | None = None):
|
||
super().__init__(parent)
|
||
self._process = QProcess(self)
|
||
self._process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
|
||
self._process.readyReadStandardOutput.connect(self._on_ready_read)
|
||
self._process.started.connect(self._on_started)
|
||
self._process.errorOccurred.connect(self._on_error)
|
||
self._process.finished.connect(self._on_finished)
|
||
self._running = False
|
||
self._starting = False
|
||
self._stop_requested = False
|
||
self._last_output_lines: deque[str] = deque(maxlen=100)
|
||
self._context: RuntimeNodeIdentity | None = None
|
||
self._attempt = 0
|
||
self._failure_reported = False
|
||
self._stdout_buffer = ""
|
||
self._secret_values: tuple[str, ...] = ()
|
||
self._compatibility_generation = 0
|
||
self._compatibility_config: dict[str, Any] | None = None
|
||
self._compatibility_relay_port = 0
|
||
self._compatibility_context: RuntimeNodeIdentity | None = None
|
||
self._chrome_fallback_pending = False
|
||
self._chrome_fallback_used = False
|
||
self._chrome_fallback_in_progress = False
|
||
self._suppress_state_change = False
|
||
self._process_generation = 0
|
||
self._config_path = HYSTERIA_CONFIG_FILE
|
||
self._last_failure_code: HysteriaFailureCode | None = None
|
||
self.diagnostic_config: dict[str, Any] | None = None
|
||
self._compatibility_allow_parallel = False
|
||
self._compatibility_verify_remote = True
|
||
self._attempt_started_at = 0.0
|
||
self._remote_authenticated = False
|
||
self.stats: dict[str, Any] = {}
|
||
self._health = BackgroundHealthCheck(self)
|
||
# A crash can leave a short-lived config behind. It is never reusable:
|
||
# every start writes a fresh one, so remove stale secrets immediately.
|
||
self._cleanup_config()
|
||
self._cleanup_stale_generation_configs()
|
||
|
||
@property
|
||
def is_running(self) -> bool:
|
||
return self._running
|
||
|
||
@property
|
||
def process_generation(self) -> int:
|
||
return self._process_generation
|
||
|
||
@property
|
||
def last_failure_code(self) -> HysteriaFailureCode | None:
|
||
return self._last_failure_code
|
||
|
||
def start(
|
||
self,
|
||
config: dict[str, Any],
|
||
relay_port: int,
|
||
*,
|
||
context: RuntimeNodeIdentity | None = None,
|
||
process_generation: int = 0,
|
||
allow_parallel: bool = False,
|
||
verify_remote: bool = True,
|
||
_compatibility_retry: bool = False,
|
||
) -> bool:
|
||
if not _compatibility_retry:
|
||
self._compatibility_generation += 1
|
||
self._chrome_fallback_pending = False
|
||
self._chrome_fallback_used = False
|
||
self._chrome_fallback_in_progress = False
|
||
self._process_generation = max(0, int(process_generation))
|
||
self._config_path = (
|
||
HYSTERIA_CONFIG_FILE.with_name(
|
||
f"{HYSTERIA_CONFIG_FILE.stem}-{self._process_generation}.json"
|
||
)
|
||
if allow_parallel and self._process_generation > 0
|
||
else HYSTERIA_CONFIG_FILE
|
||
)
|
||
self._compatibility_allow_parallel = bool(allow_parallel)
|
||
self._compatibility_verify_remote = bool(verify_remote)
|
||
self._last_failure_code = None
|
||
self._compatibility_config = deepcopy(config)
|
||
self._compatibility_relay_port = relay_port
|
||
self._compatibility_context = context
|
||
exe = HYSTERIA_PATH_DEFAULT.resolve()
|
||
if not exe.is_file():
|
||
self._begin_attempt(context, config)
|
||
self._emit_error(
|
||
f"hysteria.exe не найден: {exe}. Переустановите или обновите Zapret KVN.",
|
||
stage="validate",
|
||
)
|
||
self._clear_compatibility_state()
|
||
return False
|
||
if relay_port <= 0:
|
||
self._begin_attempt(context, config)
|
||
self._emit_error("Некорректный локальный порт Hysteria sidecar", stage="validate")
|
||
self._clear_compatibility_state()
|
||
return False
|
||
|
||
# Keep the old identity active until its process is fully stopped. This
|
||
# prevents a late finished/error signal from being attributed to the
|
||
# connection that is only about to start.
|
||
if self._process.state() != QProcess.ProcessState.NotRunning:
|
||
self._failure_reported = False
|
||
if not self.stop(expected=True, _preserve_compatibility=True):
|
||
self._emit_error(
|
||
"Не удалось остановить предыдущий процесс Hysteria",
|
||
stage="stop_previous",
|
||
)
|
||
self._clear_compatibility_state()
|
||
return False
|
||
elif self._running:
|
||
self._running = False
|
||
self.state_changed.emit(False)
|
||
|
||
self._begin_attempt(context, config)
|
||
if not allow_parallel:
|
||
self._kill_orphaned(exe)
|
||
self._cleanup_config()
|
||
temporary = self._config_path.with_suffix(".json.tmp")
|
||
try:
|
||
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
||
temporary.write_text(json.dumps(config, ensure_ascii=True, indent=2), encoding="utf-8")
|
||
temporary.chmod(0o600)
|
||
temporary.replace(self._config_path)
|
||
self.diagnostic_config = capture_runtime_config(exe, config)
|
||
except OSError as exc:
|
||
self._cleanup_config()
|
||
self._emit_error(f"Не удалось записать временный конфиг: {exc}", stage="write_config")
|
||
self._clear_compatibility_state()
|
||
return False
|
||
|
||
self._starting = True
|
||
self._stop_requested = False
|
||
self._last_output_lines.clear()
|
||
self._emit_log(
|
||
f"launch relay={PROXY_HOST}:{relay_port} remote_handshake=deferred(lazy)",
|
||
stage="spawn",
|
||
)
|
||
self._process.setWorkingDirectory(str(exe.parent))
|
||
self._process.setProgram(str(exe))
|
||
self._process.setArguments(
|
||
[
|
||
"--config",
|
||
str(self._config_path),
|
||
"--disable-update-check",
|
||
"--log-level",
|
||
"info",
|
||
"client",
|
||
]
|
||
)
|
||
self._process.start()
|
||
if not wait_for_qprocess_started(self._process, 4000):
|
||
self._starting = False
|
||
self._cleanup_config()
|
||
self._emit_error(
|
||
f"Не удалось запустить Hysteria: {self._process.errorString()}",
|
||
stage="spawn",
|
||
)
|
||
self._clear_compatibility_state()
|
||
return False
|
||
|
||
if not self._wait_until_relay_ready(relay_port):
|
||
details = self._last_output_lines[-1] if self._last_output_lines else "локальный SOCKS не открылся"
|
||
self.stop(expected=True)
|
||
self._starting = False
|
||
self._emit_error(f"Hysteria sidecar не запустился: {details}", stage="wait_ready")
|
||
return False
|
||
|
||
self._emit_log(
|
||
f"local relay ready in {int((time.monotonic() - self._attempt_started_at) * 1000)} ms",
|
||
stage="relay_ready",
|
||
)
|
||
|
||
socks = config.get("socks5")
|
||
socks = socks if isinstance(socks, dict) else {}
|
||
readiness_generation = self._compatibility_generation
|
||
if verify_remote and not self._wait_until_remote_ready(
|
||
relay_port,
|
||
username=str(socks.get("username") or ""),
|
||
password=str(socks.get("password") or ""),
|
||
):
|
||
if readiness_generation != self._compatibility_generation:
|
||
return False
|
||
details = self._last_output_lines[-1] if self._last_output_lines else "HTTPS probe через relay не завершился"
|
||
remote_failure = self._last_failure_code is not None
|
||
self.stop(expected=True)
|
||
self._starting = False
|
||
self._emit_error(
|
||
("Соединение Hysteria с сервером не установлено: " if remote_failure
|
||
else "Проверка HTTPS через Hysteria не завершилась: ") + details,
|
||
stage="remote_handshake" if remote_failure else "functional_ready",
|
||
code=self._last_failure_code
|
||
or classify_hysteria_failure(details)
|
||
or HysteriaFailureCode.TARGET_NETWORK_TIMEOUT,
|
||
)
|
||
return False
|
||
if verify_remote:
|
||
self._emit_log(
|
||
"functional readiness completed in "
|
||
f"{int((time.monotonic() - self._attempt_started_at) * 1000)} ms",
|
||
stage="functional_ready",
|
||
)
|
||
|
||
# Hysteria has parsed the config by the time its SOCKS listener is
|
||
# ready. Do not leave the URI/passwords on disk for the whole session.
|
||
self._cleanup_config()
|
||
self._starting = False
|
||
self._mark_running()
|
||
return True
|
||
|
||
def _begin_attempt(
|
||
self,
|
||
context: RuntimeNodeIdentity | None,
|
||
config: dict[str, Any],
|
||
) -> None:
|
||
self._cancel_health()
|
||
self._remote_authenticated = False
|
||
self.stats = {"remote_authenticated": False, "https_check": "pending"}
|
||
self._attempt += 1
|
||
self._attempt_started_at = time.monotonic()
|
||
self._context = context
|
||
self._failure_reported = False
|
||
self._stdout_buffer = ""
|
||
self._secret_values = self._collect_secret_values(config)
|
||
|
||
def stop(self, expected: bool = True, *, _preserve_compatibility: bool = False) -> bool:
|
||
self._cancel_health()
|
||
if not _preserve_compatibility:
|
||
self._clear_compatibility_state()
|
||
self._cleanup_config()
|
||
if self._process.state() == QProcess.ProcessState.NotRunning:
|
||
self._stop_requested = False
|
||
self._starting = False
|
||
if self._running:
|
||
self._running = False
|
||
self.state_changed.emit(False)
|
||
return True
|
||
|
||
self._stop_requested = expected
|
||
self._process.kill()
|
||
if not wait_for_qprocess_finished(self._process, 2000):
|
||
self._stop_requested = False
|
||
self._emit_error("Не удалось вовремя остановить процесс Hysteria", stage="stop")
|
||
return False
|
||
self._starting = False
|
||
return True
|
||
|
||
def _wait_until_relay_ready(self, relay_port: int, timeout: float = 10.0) -> bool:
|
||
# Shared sidecar seam: the loopback SOCKS listener must accept a TCP
|
||
# connection before the sing-box front can dial it.
|
||
return wait_for_loopback_relay(
|
||
relay_port,
|
||
timeout=timeout,
|
||
should_continue=lambda: self._process.state() != QProcess.ProcessState.NotRunning,
|
||
)
|
||
|
||
def _wait_until_remote_ready(
|
||
self,
|
||
relay_port: int,
|
||
*,
|
||
username: str,
|
||
password: str,
|
||
timeout: float = 15.0,
|
||
) -> bool:
|
||
"""Prove HTTPS egress without making one external provider authoritative."""
|
||
|
||
deadline = time.monotonic() + timeout
|
||
generation = self._compatibility_generation
|
||
failures: dict[str, str] = {}
|
||
attempts = 0
|
||
while time.monotonic() < deadline and attempts < 3:
|
||
attempts += 1
|
||
if (
|
||
self._stop_requested
|
||
or generation != self._compatibility_generation
|
||
or self._process.state() == QProcess.ProcessState.NotRunning
|
||
or self._last_failure_code in SECURITY_FAILURES
|
||
):
|
||
return False
|
||
probe_timeout = min(4.0, max(0.2, deadline - time.monotonic()))
|
||
executor = ThreadPoolExecutor(
|
||
max_workers=len(_FUNCTIONAL_HTTPS_ENDPOINTS),
|
||
thread_name_prefix="hysteria-ready",
|
||
)
|
||
futures: dict[Future[None], tuple[str, str, str]] = {
|
||
executor.submit(
|
||
self._probe_remote_endpoint,
|
||
relay_port,
|
||
username=username,
|
||
password=password,
|
||
endpoint=endpoint,
|
||
timeout=probe_timeout,
|
||
): endpoint
|
||
for endpoint in _FUNCTIONAL_HTTPS_ENDPOINTS
|
||
}
|
||
succeeded: tuple[str, str, str] | None = None
|
||
while futures and time.monotonic() < deadline:
|
||
if (
|
||
self._stop_requested
|
||
or generation != self._compatibility_generation
|
||
or self._process.state() == QProcess.ProcessState.NotRunning
|
||
or self._last_failure_code in SECURITY_FAILURES
|
||
):
|
||
executor.shutdown(wait=False, cancel_futures=True)
|
||
return False
|
||
if self._remote_authenticated:
|
||
self._monitor_health(executor, futures, failures, generation)
|
||
return True
|
||
completed = [future for future in futures if future.done()]
|
||
for future in completed:
|
||
endpoint = futures.pop(future)
|
||
error = future.exception()
|
||
if error is None:
|
||
succeeded = endpoint
|
||
break
|
||
failures[endpoint[1]] = f"{type(error).__name__}: {error}"
|
||
if succeeded is not None:
|
||
break
|
||
sleep_with_events(0.05)
|
||
if (succeeded is None and self._remote_authenticated and generation == self._compatibility_generation
|
||
and self._last_failure_code not in SECURITY_FAILURES
|
||
and self._process.state() != QProcess.ProcessState.NotRunning
|
||
and not self._stop_requested):
|
||
self._monitor_health(executor, futures, failures, generation)
|
||
return True
|
||
executor.shutdown(wait=False, cancel_futures=True)
|
||
if self._last_failure_code in SECURITY_FAILURES:
|
||
return False
|
||
if succeeded is not None:
|
||
self.stats["https_check"] = "passed"
|
||
self._emit_log(
|
||
f"functional HTTPS probe succeeded via {succeeded[1]}",
|
||
stage="functional_ready",
|
||
)
|
||
return True
|
||
sleep_with_events(0.1)
|
||
if failures:
|
||
summary = "; ".join(
|
||
f"{host}={detail}" for host, detail in sorted(failures.items())
|
||
)
|
||
self._emit_log(f"functional HTTPS probes failed: {summary}", stage="functional_ready")
|
||
return False
|
||
|
||
def _monitor_health(self, executor, futures, failures, generation):
|
||
"""Finish the existing probe wave off the startup path, without new traffic."""
|
||
self._cancel_health()
|
||
self.stats["https_check"] = "pending"
|
||
self._health.adopt(executor, {future: endpoint[1] for future, endpoint in futures.items()}, failures,
|
||
current=lambda: (generation == self._compatibility_generation
|
||
and self._running and not self._stop_requested
|
||
and self._process.state() != QProcess.ProcessState.NotRunning
|
||
and self._last_failure_code not in SECURITY_FAILURES),
|
||
complete=self._health_complete)
|
||
|
||
def _cancel_health(self):
|
||
self._health.cancel()
|
||
if self.stats.get("https_check") == "pending":
|
||
self.stats["https_check"] = "cancelled"
|
||
|
||
def _poll_health(self):
|
||
self._health.poll()
|
||
if not self._health.active and self.stats.get("https_check") == "pending":
|
||
self.stats["https_check"] = "cancelled"
|
||
|
||
def _health_complete(self, winner, failures):
|
||
if winner is not None:
|
||
self.stats["https_check"] = "passed"
|
||
self._emit_log(f"HTTPS check succeeded via {winner}", stage="health_check")
|
||
return
|
||
self.stats["https_check"] = "warning"
|
||
summary = "; ".join(f"{host}={error}" for host, error in sorted(failures.items()))
|
||
self._emit_log("authenticated server retained; HTTPS check endpoints did not respond: " + summary,
|
||
stage="health_check")
|
||
# The server handshake is already authenticated (the official client
|
||
# reported "connected to server"), so the connection is proven. The
|
||
# public DoH probe endpoints (1.1.1.1/8.8.8.8/9.9.9.9) are commonly
|
||
# blocked on censored exits, so their failure is not a user-facing
|
||
# problem. Only warn if the handshake itself was never confirmed.
|
||
if not self._remote_authenticated:
|
||
self.warning.emit("Hysteria подключена к серверу, но проверочные HTTPS-адреса не ответили. "
|
||
"Соединение сохранено; доступность сайтов пока не подтверждена.")
|
||
|
||
def _is_health_probe_error(self, line):
|
||
if not self._remote_authenticated or not (self._starting or self._health.active):
|
||
return False
|
||
parts = strip_terminal_controls(line).split("\t")
|
||
if len(parts) < 4 or parts[-2] != "SOCKS5 TCP error":
|
||
return False
|
||
try:
|
||
fields = json.loads(parts[-1])
|
||
except (ValueError, TypeError):
|
||
return False
|
||
targets = {f"{address}:443" for address, _, _ in _FUNCTIONAL_HTTPS_ENDPOINTS}
|
||
return isinstance(fields, dict) and fields.get("reqAddr") in targets
|
||
|
||
def _observe_authenticated_server(self, line):
|
||
# Parse the official CLI event before log redaction replaces JSON fields.
|
||
# A listener message or a quoted error cannot establish remote readiness.
|
||
parts = strip_terminal_controls(line).split("\t")
|
||
if (len(parts) < 4 or parts[-3:-1] != ["INFO", "connected to server"]
|
||
or not self._starting or self._stop_requested):
|
||
return
|
||
try:
|
||
fields = json.loads(parts[-1])
|
||
except (ValueError, TypeError):
|
||
return
|
||
if (not isinstance(fields, dict) or not isinstance(fields.get("addr"), str)
|
||
or not fields["addr"] or type(fields.get("count")) is not int or fields["count"] < 1):
|
||
return
|
||
self._remote_authenticated = True
|
||
self.stats["remote_authenticated"] = True
|
||
|
||
def _probe_remote_endpoint(
|
||
self,
|
||
relay_port: int,
|
||
*,
|
||
username: str,
|
||
password: str,
|
||
endpoint: tuple[str, str, str],
|
||
timeout: float,
|
||
) -> None:
|
||
probe_https(relay_port, username=username, password=password,
|
||
endpoint=endpoint, timeout=timeout)
|
||
|
||
@staticmethod
|
||
def _kill_orphaned(exe: Path) -> None:
|
||
if os.name != "nt":
|
||
return
|
||
try:
|
||
if kill_processes_by_path(exe.name, exe, timeout=5):
|
||
sleep_with_events(0.5)
|
||
except Exception:
|
||
pass
|
||
|
||
def _cleanup_config(self) -> None:
|
||
try:
|
||
self._config_path.unlink(missing_ok=True)
|
||
self._config_path.with_suffix(".json.tmp").unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
@staticmethod
|
||
def _cleanup_stale_generation_configs() -> None:
|
||
try:
|
||
for path in HYSTERIA_CONFIG_FILE.parent.glob(
|
||
f"{HYSTERIA_CONFIG_FILE.stem}-*.json*"
|
||
):
|
||
if path.is_file():
|
||
path.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
@staticmethod
|
||
def redact_log_line(line: str) -> str:
|
||
return redact_runtime_log(line)
|
||
|
||
@staticmethod
|
||
def _collect_secret_values(config: dict[str, Any]) -> tuple[str, ...]:
|
||
secret_keys = {
|
||
"auth",
|
||
"password",
|
||
"username",
|
||
"obfs-password",
|
||
"obfs_password",
|
||
"pinsha256",
|
||
"pin_sha256",
|
||
"ech",
|
||
"clientkey",
|
||
"client_key",
|
||
}
|
||
values: list[str] = []
|
||
|
||
def visit(value: Any, key: str = "") -> None:
|
||
if isinstance(value, dict):
|
||
for item_key, item in value.items():
|
||
visit(item, str(item_key).lower())
|
||
elif isinstance(value, (list, tuple)):
|
||
for item in value:
|
||
visit(item, key)
|
||
elif key in secret_keys:
|
||
text = str(value or "")
|
||
if len(text) >= 4:
|
||
values.append(text)
|
||
|
||
visit(config)
|
||
return tuple(dict.fromkeys(values))
|
||
|
||
def _format_message(self, message: str, *, stage: str) -> str:
|
||
clean = redact_runtime_log(message, secrets=self._secret_values)
|
||
fields = f"attempt={self._attempt} stage={stage}"
|
||
if self._context is not None:
|
||
fields += f" {self._context.fields()}"
|
||
return f"[hysteria][{fields}] {clean}".strip()
|
||
|
||
def _emit_log(self, message: str, *, stage: str) -> None:
|
||
formatted = self._format_message(message, stage=stage)
|
||
if not formatted:
|
||
return
|
||
self._last_output_lines.append(formatted)
|
||
self.log_received.emit(formatted)
|
||
|
||
def _emit_error(
|
||
self,
|
||
message: str,
|
||
*,
|
||
stage: str,
|
||
code: HysteriaFailureCode | None = None,
|
||
) -> None:
|
||
resolved = code or classify_hysteria_failure(message)
|
||
if resolved is None:
|
||
resolved = {
|
||
"validate": HysteriaFailureCode.LOCAL_CONFIG_INVALID,
|
||
"write_config": HysteriaFailureCode.LOCAL_CONFIG_INVALID,
|
||
"spawn": HysteriaFailureCode.LOCAL_PROCESS_START_FAILED,
|
||
"startup_exit": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
|
||
"unexpected_exit": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
|
||
"wait_ready": HysteriaFailureCode.LOCAL_RELAY_NOT_READY,
|
||
"stop": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
|
||
}.get(stage, HysteriaFailureCode.CORE_UNCLASSIFIED)
|
||
if self._failure_reported and not (
|
||
resolved in SECURITY_FAILURES and self._last_failure_code not in SECURITY_FAILURES
|
||
):
|
||
return
|
||
# A transient timeout can be logged by one parallel probe before
|
||
# another reports a definitive TLS/auth rejection. Preserve both raw
|
||
# log entries, but security must take precedence for recovery policy.
|
||
self._failure_reported = True
|
||
self._last_failure_code = resolved
|
||
formatted = self._format_message(message, stage=stage)
|
||
# The typed cause is published before generic process/state callbacks,
|
||
# so exit code 62097 cannot replace the original failure episode.
|
||
self.failure.emit(resolved.value, formatted, self._process_generation)
|
||
self.error.emit(formatted)
|
||
|
||
def _on_ready_read(self) -> None:
|
||
chunk = self._process.readAllStandardOutput()
|
||
raw = getattr(chunk, "data")()
|
||
text = decode_output(bytes(raw)) if isinstance(raw, (bytes, bytearray)) else str(raw)
|
||
self._stdout_buffer += text
|
||
lines = self._stdout_buffer.splitlines(keepends=True)
|
||
self._stdout_buffer = ""
|
||
for item in lines:
|
||
if not item.endswith(("\n", "\r")):
|
||
self._stdout_buffer = item
|
||
continue
|
||
self._emit_process_line(item.rstrip("\r\n"))
|
||
|
||
def _emit_process_line(self, line: str) -> None:
|
||
self._observe_authenticated_server(line)
|
||
from ...diagnostics.runtime_errors import classify_core_error
|
||
clean = redact_runtime_log(line, secrets=self._secret_values)
|
||
_, action = classify_core_error(clean)
|
||
if self._is_health_probe_error(line) and classify_hysteria_failure(clean) not in SECURITY_FAILURES:
|
||
self._emit_log(clean, stage="health_check")
|
||
return
|
||
if action == "record_only":
|
||
# A loopback SOCKS client/probe closing its socket is not a remote
|
||
# server failure. Preserve evidence without spending recovery.
|
||
self._emit_log(clean, stage="local_client")
|
||
return
|
||
lowered = clean.lower()
|
||
stage = (
|
||
"remote_handshake"
|
||
if any(
|
||
marker in lowered
|
||
for marker in (
|
||
"crypto_error",
|
||
"no recent network activity",
|
||
"handshake",
|
||
"failed to initialize client",
|
||
"tls:",
|
||
"certificate",
|
||
"pinned",
|
||
"x509",
|
||
"authentication failed",
|
||
"server rejected",
|
||
"connection refused",
|
||
"actively refused",
|
||
"forcibly closed",
|
||
)
|
||
)
|
||
else "runtime"
|
||
)
|
||
if clean:
|
||
self._emit_log(clean, stage=stage)
|
||
if self._is_chrome_parrot_compatibility_error(clean):
|
||
quic = (self._compatibility_config or {}).get("quic", {})
|
||
if quic.get("disableChromeParrot"):
|
||
self._emit_error(clean, stage="remote_handshake")
|
||
elif not self._chrome_fallback_used and not self._chrome_fallback_pending:
|
||
self._schedule_chrome_parrot_fallback()
|
||
elif (
|
||
self._chrome_fallback_used
|
||
and not self._chrome_fallback_pending
|
||
and not self._chrome_fallback_in_progress
|
||
):
|
||
self._emit_error(clean, stage="remote_handshake")
|
||
else:
|
||
failure = classify_hysteria_failure(clean)
|
||
if failure is not None:
|
||
if self._starting and failure not in SECURITY_FAILURES:
|
||
# One failed probe is not the verdict of the whole start.
|
||
# Keep the cause and raw log until the readiness gate ends.
|
||
if self._last_failure_code not in SECURITY_FAILURES:
|
||
self._last_failure_code = failure
|
||
else:
|
||
self._emit_error(clean, stage=stage, code=failure)
|
||
|
||
@staticmethod
|
||
def _is_chrome_parrot_compatibility_error(line: str) -> bool:
|
||
lowered = str(line or "").lower()
|
||
return (
|
||
"crypto_error 0x150" in lowered
|
||
and "(remote)" in lowered
|
||
and "tls: internal error" in lowered
|
||
)
|
||
|
||
def _schedule_chrome_parrot_fallback(self) -> None:
|
||
if self._compatibility_config is None or self._compatibility_relay_port <= 0:
|
||
return
|
||
self._chrome_fallback_pending = True
|
||
self._chrome_fallback_used = True
|
||
generation = self._compatibility_generation
|
||
self._emit_log(
|
||
"remote TLS internal_error; scheduling one compatibility retry "
|
||
"without Chrome QUIC parroting",
|
||
stage="compatibility_retry",
|
||
)
|
||
QTimer.singleShot(0, lambda: self._run_chrome_parrot_fallback(generation))
|
||
|
||
def _run_chrome_parrot_fallback(self, generation: int) -> None:
|
||
if generation != self._compatibility_generation:
|
||
return
|
||
self._chrome_fallback_pending = False
|
||
if (
|
||
self._compatibility_config is None
|
||
or self._compatibility_relay_port <= 0
|
||
or self._stop_requested
|
||
):
|
||
return
|
||
|
||
config = deepcopy(self._compatibility_config)
|
||
quic = config.get("quic")
|
||
if not isinstance(quic, dict):
|
||
quic = {}
|
||
config["quic"] = quic
|
||
quic["disableChromeParrot"] = True
|
||
relay_port = self._compatibility_relay_port
|
||
context = self._compatibility_context
|
||
self._chrome_fallback_in_progress = True
|
||
self._suppress_state_change = True
|
||
try:
|
||
started = self.start(
|
||
config,
|
||
relay_port,
|
||
context=context,
|
||
process_generation=self._process_generation,
|
||
allow_parallel=self._compatibility_allow_parallel,
|
||
verify_remote=self._compatibility_verify_remote,
|
||
_compatibility_retry=True,
|
||
)
|
||
finally:
|
||
self._suppress_state_change = False
|
||
self._chrome_fallback_in_progress = False
|
||
if not started:
|
||
if not self._running:
|
||
self.state_changed.emit(False)
|
||
self._emit_error(
|
||
"Не удалось повторно запустить Hysteria2 в режиме совместимости сертификата.",
|
||
stage="compatibility_retry",
|
||
)
|
||
self._clear_compatibility_state()
|
||
|
||
def _clear_compatibility_state(self) -> None:
|
||
self._compatibility_generation += 1
|
||
self._compatibility_config = None
|
||
self._compatibility_relay_port = 0
|
||
self._compatibility_context = None
|
||
self._chrome_fallback_pending = False
|
||
self._chrome_fallback_used = False
|
||
self._chrome_fallback_in_progress = False
|
||
|
||
def _flush_stdout_buffer(self) -> None:
|
||
self._on_ready_read()
|
||
if self._stdout_buffer:
|
||
tail = self._stdout_buffer
|
||
self._stdout_buffer = ""
|
||
self._emit_process_line(tail)
|
||
|
||
def _on_started(self) -> None:
|
||
# Readiness is established by the SOCKS probe in start(); do not expose
|
||
# the process as a healthy sidecar merely because CreateProcess worked.
|
||
self._emit_log("process started; waiting for local SOCKS relay", stage="process_started")
|
||
self.started.emit()
|
||
|
||
def _mark_running(self) -> None:
|
||
if self._running:
|
||
return
|
||
self._running = True
|
||
self._last_failure_code = None
|
||
self._failure_reported = False
|
||
self._emit_log(
|
||
("local SOCKS relay and authenticated Hysteria server are ready"
|
||
if self._remote_authenticated else "local SOCKS relay and functional HTTPS handshake are ready"),
|
||
stage="remote_ready" if self._remote_authenticated else "functional_ready",
|
||
)
|
||
if not self._suppress_state_change:
|
||
self.state_changed.emit(True)
|
||
|
||
def _on_error(self, process_error: QProcess.ProcessError) -> None:
|
||
if self._stop_requested and process_error == QProcess.ProcessError.Crashed:
|
||
return
|
||
self._emit_error(
|
||
f"Ошибка процесса Hysteria: {process_error.name} ({self._process.errorString()})",
|
||
stage="spawn" if self._starting else "process_error",
|
||
)
|
||
|
||
def _on_finished(self, exit_code: int, _status: QProcess.ExitStatus) -> None:
|
||
self._cancel_health()
|
||
was_running = self._running
|
||
was_starting = self._starting
|
||
expected = self._stop_requested
|
||
self._flush_stdout_buffer()
|
||
self._cleanup_config()
|
||
self._running = False
|
||
self._starting = False
|
||
self._stop_requested = False
|
||
compatibility_pending = self._chrome_fallback_pending
|
||
if not expected and not compatibility_pending:
|
||
details = self._last_output_lines[-1] if self._last_output_lines else "без диагностического сообщения"
|
||
self._emit_error(
|
||
f"Hysteria неожиданно завершилась (код {exit_code}): {details}",
|
||
stage="startup_exit" if was_starting else "unexpected_exit",
|
||
code=self._last_failure_code or classify_hysteria_failure(details, process_exited=True),
|
||
)
|
||
self._clear_compatibility_state()
|
||
elif not expected:
|
||
self._emit_log(
|
||
f"process exited with code {exit_code}; continuing scheduled compatibility retry",
|
||
stage="compatibility_retry",
|
||
)
|
||
if was_running and not self._suppress_state_change and not compatibility_pending:
|
||
self.state_changed.emit(False)
|
||
self.stopped.emit(exit_code)
|