zapret-kvn/xray_fluent/network/connectivity_test.py
loop-uh 70ac787146
Some checks failed
Windows project source guards / test (push) Has been cancelled
Проверки не выдают серверы: рукопожатие Chrome, имя сервера не уходит в открытый DNS
Требование незаметности: ни проверки, ни подключение не должны выдавать
провайдеру (ТСПУ) VPN-серверы.

- Без fp в ссылке рукопожатие с сервером — uTLS chrome, а не ClientHello
  библиотеки Go, который отличим с первого пакета. Значение из ссылки
  сохраняется, QUIC (HTTP/3, mKCP) не трогаем, хранимый узел не меняется —
  отпечаток подставляется во все конфиги ядра (sing-box, Xray, пул, замер).
- Фоновое обновление адреса больше не спрашивает системный резолвер про
  имя сервера при каждом подключении. Подмена DNS выводится только из
  запроса, который всё равно понадобился (DoH недоступен, провайдер говорит
  «имени нет», а кеш его знает).
- После подключения имена всех серверов пула больше не резолвятся (DoH к
  пяти резолверам на каждое имя) ради маскировки журнала: регистрируются
  только адреса из кеша; цель горячего переключения резолвится перед ним.
- Подписка по http:// к адресу в интернете грузится только через туннель:
  напрямую провайдер увидел бы весь список серверов; редирект на http://
  на прямом пути отклоняется. Локальная сеть — как раньше.
- Проверка связи — браузерный User-Agent вместо ZapretKVN/0.4.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 17:57:00 +03:00

45 lines
1.5 KiB
Python

from __future__ import annotations
import time
from urllib.request import ProxyHandler, Request
from .http_utils import build_opener
from .speed_test_worker import SPEED_TEST_USER_AGENT
from PyQt6.QtCore import QThread, pyqtSignal
class ConnectivityTestWorker(QThread):
result = pyqtSignal(bool, str, object)
def __init__(self, http_port: int, url: str, timeout: float = 8.0, tun_mode: bool = False):
super().__init__()
self._http_port = http_port
self._url = url
self._timeout = timeout
self._tun_mode = tun_mode
def run(self) -> None:
if self._tun_mode:
opener = build_opener()
else:
proxy_url = f"http://127.0.0.1:{self._http_port}"
opener = build_opener(
ProxyHandler(
{
"http": proxy_url,
"https": proxy_url,
}
)
)
request = Request(self._url, headers={"User-Agent": SPEED_TEST_USER_AGENT})
started = time.perf_counter()
try:
with opener.open(request, timeout=self._timeout) as response:
response.read(16)
status_code = getattr(response, "status", 200)
elapsed_ms = int((time.perf_counter() - started) * 1000)
self.result.emit(True, f"{self._url} -> HTTP {status_code}", elapsed_ms)
except Exception as exc:
self.result.emit(False, f"{self._url} -> {exc}", None)